Finance-Data-Engine/app.py at main · Datamathican/Finance-Data-Engine · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import streamlit as st
import pandas as pd
import plotly.graph_objs as go
from sklearn.metrics import mean_squared_error
from math import sqrt

# Import modules
from src.data_ingestion import get_natwest_data
from src.governance import validate_data
from src.models import train_prophet, train_arima, train_holtwinters

st.set_page_config(page_title="TreasuryGuard", layout="wide")

st.title("TreasuryGuard: Liquidity Stress Testing Engine")
st.markdown("### Champion/Challenger Framework for Liquidity Forecasting")

# 1. Ingest
with st.spinner("Ingesting Real-Time Market Data..."):
    df = get_natwest_data()

# 2. Split Data
TEST_DAYS = 30
train = df.iloc[:-TEST_DAYS]
test = df.iloc[-TEST_DAYS:]

# 3. Governance
if validate_data(df):
    st.sidebar.success(" BCBS 239 Data Governance Passed")

    # 4. Train Models (The Tournament)
    if st.button("Run Model Tournament"):
        with st.spinner("Training Prophet, ARIMA, and Holt-Winters..."):

            # Get Predictions
            pred_p, full_forecast_p = train_prophet(train, TEST_DAYS)
            pred_a = train_arima(train, TEST_DAYS)
            pred_h = train_holtwinters(train, TEST_DAYS)

            # Evaluate (RMSE)
            rmse_p = sqrt(mean_squared_error(test['y'], pred_p))
            rmse_a = sqrt(mean_squared_error(test['y'], pred_a))
            rmse_h = sqrt(mean_squared_error(test['y'], pred_h))

            # Winner Logic
            scores = {"Prophet": rmse_p, "ARIMA": rmse_a, "Holt-Winters": rmse_h}
            winner = min(scores, key=scores.get)

            st.success(f" The Champion Model is: **{winner}** (Lowest RMSE)")

            # 5. Visualization
            fig = go.Figure()
            # Truth
            fig.add_trace(go.Scatter(x=test['ds'], y=test['y'], mode='lines+markers', name='Actual Data', line=dict(color='black', width=2)))
            # Models
            fig.add_trace(go.Scatter(x=test['ds'], y=pred_p, mode='lines', name=f'Prophet (RMSE: {rmse_p:.0f})'))
            fig.add_trace(go.Scatter(x=test['ds'], y=pred_a, mode='lines', name=f'ARIMA (RMSE: {rmse_a:.0f})', line=dict(dash='dot')))
            fig.add_trace(go.Scatter(x=test['ds'], y=pred_h, mode='lines', name=f'Holt-Winters (RMSE: {rmse_h:.0f})', line=dict(dash='dash')))

            st.plotly_chart(fig, use_container_width=True)

            # 6. Stress Testing Section
            st.markdown("---")
            st.subheader(" Interactive Stress Testing")
            shock = st.slider("Select Liquidity Shock (%)", 0, 50, 0)

            if shock > 0:
                st.warning(f"Simulating {shock}% Market Crash Scenario on Champion Model ({winner})")

                # Apply stress to Prophet forecast for visualization
                stressed_forecast = full_forecast_p.iloc[-TEST_DAYS:]['yhat'] * (1 - shock/100)

                stress_fig = go.Figure()
                stress_fig.add_trace(go.Scatter(x=test['ds'], y=test['y'], name='Actuals', line=dict(color='gray')))
                stress_fig.add_trace(go.Scatter(x=test['ds'], y=pred_p, name='Baseline Forecast', line=dict(color='blue')))
                stress_fig.add_trace(go.Scatter(x=test['ds'], y=stressed_forecast, name='Stressed Scenario', line=dict(color='red', width=3)))

                st.plotly_chart(stress_fig, use_container_width=True)

else:
    st.error(" Pipeline Halted: Data Quality Breach Detected.")