""", unsafe_allow_html=True)
- # ββ Per-ETF rows ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
for i, (name, score) in enumerate(sorted_pairs):
- is_winner = (name == best_name)
- is_last = (i == len(sorted_pairs) - 1)
- bar_w = int(score / max_score * 100)
+ is_winner = (name == best_name)
+ is_last = (i == len(sorted_pairs) - 1)
+ bar_w = int(score / max_score * 100)
name_style = "font-weight:700; color:#00897b;" if is_winner else "color:#444;"
bar_color = color if is_winner else "#b2dfdb" if score > max_score * 0.5 else "#e0e0e0"
star = " β
" if is_winner else ""
@@ -134,10 +146,9 @@ def show_conviction_panel(conviction: dict):
box-shadow:0 2px 8px rgba(0,0,0,0.07);">
{name}{star}
-
@@ -145,7 +156,7 @@ def show_conviction_panel(conviction: dict):
""", unsafe_allow_html=True)
st.caption(
- "Z-score = std deviations the top ETF's probability sits above the mean of all ETF probabilities. "
+ "Z-score = std deviations the top ETF's probability sits above the mean. "
"Higher β model is more decisive."
)
@@ -153,60 +164,44 @@ def show_conviction_panel(conviction: dict):
# ββ Metrics row βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def show_metrics_row(result: dict, tbill_rate: float):
- """Five-column metric display."""
col1, col2, col3, col4, col5 = st.columns(5)
-
- col1.metric(
- "π Annualised Return",
- f"{result['ann_return']*100:.2f}%",
- delta=f"vs T-bill: {(result['ann_return'] - tbill_rate)*100:.2f}%",
- )
- col2.metric(
- "π Sharpe Ratio",
- f"{result['sharpe']:.2f}",
- delta="Risk-Adjusted" if result['sharpe'] > 1 else "Below Threshold",
- )
- col3.metric(
- "π― Hit Ratio (15d)",
- f"{result['hit_ratio']*100:.0f}%",
- delta="Strong" if result['hit_ratio'] > 0.6 else "Weak",
- )
- col4.metric(
- "π Max Drawdown",
- f"{result['max_dd']*100:.2f}%",
- delta="Peak to Trough",
- )
- col5.metric(
- "β οΈ Max Daily DD",
- f"{result['max_daily_dd']*100:.2f}%",
- delta="Worst Day",
- )
+ col1.metric("π Ann. Return", f"{result['ann_return']*100:.2f}%",
+ delta=f"vs T-bill: {(result['ann_return'] - tbill_rate)*100:.2f}%")
+ col2.metric("π Sharpe", f"{result['sharpe']:.2f}",
+ delta="Strong" if result['sharpe'] > 1 else "Weak")
+ col3.metric("π― Hit Ratio 15d", f"{result['hit_ratio']*100:.0f}%",
+ delta="Good" if result['hit_ratio'] > 0.55 else "Weak")
+ col4.metric("π Max Drawdown", f"{result['max_dd']*100:.2f}%",
+ delta="Peak to Trough")
+ col5.metric("β οΈ Max Daily DD", f"{result['max_daily_dd']*100:.2f}%",
+ delta="Worst Day")
# ββ Comparison table ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def show_comparison_table(comparison_df: pd.DataFrame):
- """Styled comparison table for all three approaches."""
def highlight_winner(row):
if "WINNER" in str(row.get("Winner", "")):
return ["background-color: rgba(0,200,150,0.15); font-weight:bold"] * len(row)
return [""] * len(row)
- styled = comparison_df.style.apply(highlight_winner, axis=1).set_properties(**{
- "text-align": "center",
- "font-size": "14px",
- }).set_table_styles([
- {"selector": "th", "props": [("font-size", "14px"), ("font-weight", "bold"),
- ("text-align", "center")]},
- {"selector": "td", "props": [("padding", "10px")]},
- ])
+ styled = (
+ comparison_df.style
+ .apply(highlight_winner, axis=1)
+ .set_properties(**{"text-align": "center", "font-size": "14px"})
+ .set_table_styles([
+ {"selector": "th", "props": [("font-size", "14px"),
+ ("font-weight", "bold"),
+ ("text-align", "center")]},
+ {"selector": "td", "props": [("padding", "10px")]},
+ ])
+ )
st.dataframe(styled, use_container_width=True)
# ββ Audit trail βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def show_audit_trail(audit_trail: list):
- """Last 20 days styled audit trail."""
if not audit_trail:
st.info("No audit trail data available.")
return
@@ -214,59 +209,19 @@ def show_audit_trail(audit_trail: list):
df = pd.DataFrame(audit_trail).tail(20)[["Date", "Signal", "Net_Return"]]
def color_return(val):
- return "color: #00c896; font-weight:bold" if val > 0 else "color: #ff4b4b; font-weight:bold"
-
- styled = df.style.applymap(color_return, subset=["Net_Return"]).format(
- {"Net_Return": "{:.2%}"}
- ).set_properties(**{
- "font-size": "16px",
- "text-align": "center",
- }).set_table_styles([
- {"selector": "th", "props": [("font-size", "16px"), ("font-weight", "bold"),
- ("text-align", "center")]},
- {"selector": "td", "props": [("padding", "10px")]},
- ])
+ return ("color: #00c896; font-weight:bold" if val > 0
+ else "color: #ff4b4b; font-weight:bold")
+
+ styled = (
+ df.style
+ .map(color_return, subset=["Net_Return"])
+ .format({"Net_Return": "{:.2%}"})
+ .set_properties(**{"font-size": "14px", "text-align": "center"})
+ .set_table_styles([
+ {"selector": "th", "props": [("font-size", "14px"),
+ ("font-weight", "bold"),
+ ("text-align", "center")]},
+ {"selector": "td", "props": [("padding", "10px")]},
+ ])
+ )
st.dataframe(styled, use_container_width=True, height=500)
-
-
-# ββ All models' next day signals panel βββββββββββββββββββββββββββββββββββββββ
-
-def show_all_signals_panel(all_signals: dict, target_etfs: list, include_cash: bool, next_date):
- """
- Compact panel showing what each model predicts for next trading day,
- with top probability displayed.
- """
- APPROACH_COLORS = {
- "Approach 1": "#00ffc8",
- "Approach 2": "#7c6aff",
- "Approach 3": "#ff6b6b",
- }
-
- st.subheader(f"ποΈ All Models β {next_date.strftime('%Y-%m-%d')} Signals")
-
- cols = st.columns(len(all_signals))
- for col, (name, info) in zip(cols, all_signals.items()):
- color = APPROACH_COLORS.get(name, "#888888")
- signal = info["signal"]
- proba = info["proba"]
- top_prob = float(np.max(proba)) * 100
- is_winner = info["is_winner"]
- border = f"3px solid {color}"
- badge = " β WINNER" if is_winner else ""
-
- col.markdown(f"""
-
-
- {name.upper()}{badge}
-
-
- {signal}
-
-
- Top prob: {top_prob:.1f}%
-
-
- """, unsafe_allow_html=True)
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py
index fc3b824bf2f3cbedd7b956d6969fad5e5ecca980..3965d4d7256f2fd352e4ec58bb1b30bfb2c3fb5c 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py
@@ -1,22 +1,15 @@
"""
ui/charts.py
-All Plotly chart builders for the Streamlit UI.
+Plotly chart builders.
+Equity curve: winner + SPY + AGG only. Y-axis as % growth (not raw multiplier).
"""
import numpy as np
import pandas as pd
import plotly.graph_objects as go
-
-APPROACH_COLOURS = {
- "Approach 1": "#00ffc8",
- "Approach 2": "#7c6aff",
- "Approach 3": "#ff6b6b",
-}
-BENCHMARK_COLOURS = {
- "SPY": "#ff4b4b",
- "AGG": "#ffa500",
-}
+WINNER_COLOUR = "#00ffc8"
+BENCHMARK_COLOURS = {"SPY": "#ff4b4b", "AGG": "#ffa500"}
def equity_curve_chart(
@@ -28,117 +21,72 @@ def equity_curve_chart(
tbill_rate: float,
) -> go.Figure:
"""
- Equity curve chart showing all three approaches + SPY + AGG benchmarks.
-
- Args:
- results : {approach_name: result_dict}
- winner_name : highlighted approach
- plot_dates : DatetimeIndex for x-axis
- df : full DataFrame (for benchmark columns)
- test_slice : slice object to extract test-period benchmark returns
- tbill_rate : for benchmark metric calculation
+ Equity curve: winner strategy vs SPY and AGG.
+ Y-axis shows % growth (cum_return - 1) * 100 for readability.
"""
from strategy.backtest import compute_benchmark_metrics
fig = go.Figure()
- # ββ Strategy lines ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
- for name, res in results.items():
- if res is None:
- continue
- colour = APPROACH_COLOURS.get(name, "#aaaaaa")
- width = 3 if name == winner_name else 1.5
- dash = "solid" if name == winner_name else "dot"
-
- n = min(len(res["cum_returns"]), len(plot_dates))
-
+ # ββ Winner strategy βββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+ winner_res = results.get(winner_name)
+ if winner_res is not None:
+ cum = winner_res["cum_returns"]
+ n = min(len(cum), len(plot_dates))
fig.add_trace(go.Scatter(
x=plot_dates[:n],
- y=res["cum_returns"][:n],
+ y=(cum[:n] - 1) * 100,
mode="lines",
- name=f"{name} {'β
' if name == winner_name else ''}",
- line=dict(color=colour, width=width, dash=dash),
- fill="tozeroy" if name == winner_name else None,
- fillcolor=f"rgba({_hex_to_rgb(colour)},0.07)" if name == winner_name else None,
+ name=f"{winner_name} β
",
+ line=dict(color=WINNER_COLOUR, width=2.5),
+ fill="tozeroy",
+ fillcolor="rgba(0,255,200,0.07)",
))
- # ββ Benchmark: SPY ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+ # ββ SPY benchmark βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
if "SPY_Ret" in df.columns:
- spy_rets = df["SPY_Ret"].iloc[test_slice].values
+ spy_rets = df["SPY_Ret"].iloc[test_slice].values.copy()
+ spy_rets = np.clip(spy_rets, -0.5, 0.5) # sanity clip
+ spy_rets = spy_rets[~np.isnan(spy_rets)]
n = min(len(spy_rets), len(plot_dates))
spy_m = compute_benchmark_metrics(spy_rets[:n], tbill_rate)
fig.add_trace(go.Scatter(
x=plot_dates[:n],
- y=spy_m["cum_returns"],
+ y=(spy_m["cum_returns"] - 1) * 100,
mode="lines",
- name="SPY (Equity BM)",
+ name="SPY",
line=dict(color=BENCHMARK_COLOURS["SPY"], width=1.5, dash="dot"),
))
- # ββ Benchmark: AGG ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+ # ββ AGG benchmark βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
if "AGG_Ret" in df.columns:
- agg_rets = df["AGG_Ret"].iloc[test_slice].values
+ agg_rets = df["AGG_Ret"].iloc[test_slice].values.copy()
+ agg_rets = np.clip(agg_rets, -0.5, 0.5)
+ agg_rets = agg_rets[~np.isnan(agg_rets)]
n = min(len(agg_rets), len(plot_dates))
agg_m = compute_benchmark_metrics(agg_rets[:n], tbill_rate)
fig.add_trace(go.Scatter(
x=plot_dates[:n],
- y=agg_m["cum_returns"],
+ y=(agg_m["cum_returns"] - 1) * 100,
mode="lines",
- name="AGG (Bond BM)",
+ name="AGG",
line=dict(color=BENCHMARK_COLOURS["AGG"], width=1.5, dash="dot"),
))
fig.update_layout(
template="plotly_dark",
- height=460,
+ height=420,
hovermode="x unified",
- showlegend=True,
legend=dict(yanchor="top", y=0.99, xanchor="left", x=0.01, font=dict(size=11)),
xaxis_title="Date",
- yaxis_title="Cumulative Return (Γ)",
+ yaxis_title="Cumulative Return (%)",
margin=dict(l=50, r=30, t=20, b=50),
+ yaxis=dict(ticksuffix="%"),
)
return fig
-def comparison_bar_chart(results: dict, winner_name: str) -> go.Figure:
- """
- Horizontal bar chart comparing annualised returns across all three approaches.
- """
- names = []
- returns = []
- colours = []
-
- for name, res in results.items():
- if res is None:
- continue
- names.append(name)
- returns.append(res["ann_return"] * 100)
- colours.append(APPROACH_COLOURS.get(name, "#aaaaaa"))
-
- fig = go.Figure(go.Bar(
- x=returns,
- y=names,
- orientation="h",
- marker_color=colours,
- text=[f"{r:.1f}%" for r in returns],
- textposition="auto",
- ))
-
- fig.update_layout(
- template="plotly_dark",
- height=200,
- xaxis_title="Annualised Return (%)",
- margin=dict(l=100, r=30, t=10, b=40),
- showlegend=False,
- )
- return fig
-
-
-# ββ Helper ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
def _hex_to_rgb(hex_color: str) -> str:
- """Convert #rrggbb to 'r,g,b' string for rgba()."""
h = hex_color.lstrip("#")
r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
return f"{r},{g},{b}"
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
index b8f6d0445d0bdd6c7d364b1c63d52cc7c37f8143..b020ba27880c8a58a313304eab320de88361dd57 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
@@ -1,28 +1,53 @@
"""
models/base.py
-Shared utilities for all three CNN-LSTM variants.
-Key fix: class_weight support to prevent majority-class collapse.
+Shared utilities for all CNN-LSTM variants.
+Optimised for CPU training on HF Spaces.
"""
import numpy as np
-import pandas as pd
+import hashlib
+import pickle
+import os
+from pathlib import Path
from sklearn.preprocessing import RobustScaler
from sklearn.utils.class_weight import compute_class_weight
-import tensorflow as tf
-from tensorflow import keras
-SEED = 42
-tf.random.set_seed(SEED)
+SEED = 42
+CACHE_DIR = Path("/tmp/p2_model_cache")
+CACHE_DIR.mkdir(exist_ok=True)
+
np.random.seed(SEED)
+# ββ Cache helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def make_cache_key(last_date: str, start_yr: int, fee_bps: int,
+ epochs: int, split: str, include_cash: bool,
+ lookback: int) -> str:
+ raw = f"{last_date}_{start_yr}_{fee_bps}_{epochs}_{split}_{include_cash}_{lookback}"
+ return hashlib.md5(raw.encode()).hexdigest()
+
+
+def save_cache(key: str, payload: dict):
+ path = CACHE_DIR / f"{key}.pkl"
+ with open(path, "wb") as f:
+ pickle.dump(payload, f)
+
+
+def load_cache(key: str) -> dict | None:
+ path = CACHE_DIR / f"{key}.pkl"
+ if path.exists():
+ try:
+ with open(path, "rb") as f:
+ return pickle.load(f)
+ except Exception:
+ path.unlink(missing_ok=True)
+ return None
+
+
# ββ Sequence builder ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def build_sequences(features: np.ndarray, targets: np.ndarray, lookback: int):
- """
- Build supervised sequences for CNN-LSTM input.
- X[i] = features[i : i+lookback] β predicts y[i+lookback]
- """
X, y = [], []
for i in range(lookback, len(features)):
X.append(features[i - lookback: i])
@@ -36,35 +61,25 @@ def train_val_test_split(X, y, train_pct=0.70, val_pct=0.15):
n = len(X)
t1 = int(n * train_pct)
t2 = int(n * (train_pct + val_pct))
- return (
- X[:t1], y[:t1],
- X[t1:t2], y[t1:t2],
- X[t2:], y[t2:],
- )
+ return X[:t1], y[:t1], X[t1:t2], y[t1:t2], X[t2:], y[t2:]
# ββ Feature scaling βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def scale_features(X_train, X_val, X_test):
- n_feat = X_train.shape[2]
- scaler = RobustScaler()
+ n_feat = X_train.shape[2]
+ scaler = RobustScaler()
scaler.fit(X_train.reshape(-1, n_feat))
-
def _t(X):
s = X.shape
return scaler.transform(X.reshape(-1, n_feat)).reshape(s)
-
return _t(X_train), _t(X_val), _t(X_test), scaler
# ββ Label builder βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def returns_to_labels(y_raw, include_cash=True, cash_threshold=0.0):
- """
- Assign label = argmax(returns).
- If include_cash and best return < cash_threshold β label = n_etfs (CASH).
- """
- best = np.argmax(y_raw, axis=1)
+ best = np.argmax(y_raw, axis=1)
if include_cash:
best_ret = y_raw[np.arange(len(y_raw)), best]
cash_idx = y_raw.shape[1]
@@ -77,35 +92,22 @@ def returns_to_labels(y_raw, include_cash=True, cash_threshold=0.0):
# ββ Class weights βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def compute_class_weights(y_labels: np.ndarray, n_classes: int) -> dict:
- """
- Compute balanced class weights to counteract majority-class collapse.
- Returns dict {class_index: weight} for use in model.fit().
- """
- classes = np.arange(n_classes)
present = np.unique(y_labels)
-
try:
- weights = compute_class_weight(
- class_weight="balanced",
- classes=present,
- y=y_labels,
- )
+ weights = compute_class_weight("balanced", classes=present, y=y_labels)
weight_dict = {int(c): float(w) for c, w in zip(present, weights)}
except Exception:
weight_dict = {}
-
- # Fill any missing classes with weight 1.0
- for c in classes:
+ for c in range(n_classes):
if c not in weight_dict:
weight_dict[c] = 1.0
-
return weight_dict
# ββ Callbacks βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-def get_callbacks(patience_es=20, patience_lr=10, min_lr=1e-6):
- """Longer patience to allow models time to learn past majority class."""
+def get_callbacks(patience_es=15, patience_lr=8, min_lr=1e-6):
+ from tensorflow import keras
return [
keras.callbacks.EarlyStopping(
monitor="val_loss",
@@ -123,51 +125,76 @@ def get_callbacks(patience_es=20, patience_lr=10, min_lr=1e-6):
]
-# ββ Output head βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+# ββ Lightweight output head (CPU-optimised) βββββββββββββββββββββββββββββββββββ
def classification_head(x, n_classes: int, dropout: float = 0.3):
- x = keras.layers.Dense(64, activation="relu")(x)
- x = keras.layers.BatchNormalization()(x)
- x = keras.layers.Dropout(dropout)(x)
+ """Smaller head than original β faster on CPU, less overfitting risk."""
+ from tensorflow import keras
x = keras.layers.Dense(32, activation="relu")(x)
- x = keras.layers.Dropout(dropout / 2)(x)
+ x = keras.layers.Dropout(dropout)(x)
x = keras.layers.Dense(n_classes, activation="softmax")(x)
return x
-# ββ Prediction ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def predict_classes(model, X_test: np.ndarray) -> tuple:
- proba = model.predict(X_test, verbose=0)
- return np.argmax(proba, axis=1), proba
-
+# ββ Auto lookback selection βββββββββββββββββββββββββββββββββββββββββββββββββββ
-# ββ Metrics βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def evaluate_returns(
- preds, proba, y_raw_test, target_etfs, tbill_rate, fee_bps, include_cash=True,
+def find_best_lookback(
+ X_raw: np.ndarray,
+ y_raw: np.ndarray,
+ y_labels_fn,
+ train_pct: float,
+ val_pct: float,
+ n_classes: int,
+ include_cash: bool,
+ candidates: list = None,
):
- n_etfs = len(target_etfs)
- daily_tbill = tbill_rate / 252
- strat_rets = []
-
- for i, cls in enumerate(preds):
- if include_cash and cls == n_etfs:
- net = daily_tbill - fee_bps / 10000
- else:
- cls = min(int(cls), n_etfs - 1)
- net = float(y_raw_test[i][cls]) - fee_bps / 10000
- strat_rets.append(net)
-
- strat_rets = np.array(strat_rets)
- cum_returns = np.cumprod(1 + strat_rets)
- ann_return = cum_returns[-1] ** (252 / len(strat_rets)) - 1
-
- last_proba = proba[-1]
- next_cls = int(np.argmax(last_proba))
- next_etf = (
- "CASH" if (include_cash and next_cls == n_etfs)
- else target_etfs[min(next_cls, n_etfs - 1)].replace("_Ret", "")
- )
-
- return strat_rets, ann_return, cum_returns, last_proba, next_etf
+ """
+ Train a fast lightweight CNN on each lookback candidate using val loss.
+ Returns best lookback int.
+ Uses only Approach 1 architecture (fastest) to pick the winner.
+ """
+ from tensorflow import keras
+
+ if candidates is None:
+ candidates = [30, 45, 60]
+
+ best_lb = candidates[0]
+ best_loss = np.inf
+
+ for lb in candidates:
+ try:
+ X_seq, y_seq = build_sequences(X_raw, y_raw, lb)
+ y_lab = y_labels_fn(y_seq)
+
+ X_tr, y_tr, X_v, y_v, _, _ = train_val_test_split(X_seq, y_lab, train_pct, val_pct)
+ X_tr_s, X_v_s, _, _ = scale_features(X_tr, X_v, X_v)
+
+ cw = compute_class_weights(y_tr, n_classes)
+
+ # Tiny fast model just for lookback selection
+ inp = keras.Input(shape=X_tr_s.shape[1:])
+ x = keras.layers.Conv1D(16, min(3, lb), padding="causal", activation="relu")(inp)
+ x = keras.layers.GlobalAveragePooling1D()(x)
+ out = keras.layers.Dense(n_classes, activation="softmax")(x)
+ m = keras.Model(inp, out)
+ m.compile(optimizer="adam", loss="sparse_categorical_crossentropy")
+
+ hist = m.fit(
+ X_tr_s, y_tr,
+ validation_data=(X_v_s, y_v),
+ epochs=15,
+ batch_size=64,
+ class_weight=cw,
+ callbacks=[keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True)],
+ verbose=0,
+ )
+ val_loss = min(hist.history.get("val_loss", [np.inf]))
+ if val_loss < best_loss:
+ best_loss = val_loss
+ best_lb = lb
+
+ del m
+ except Exception:
+ continue
+
+ return best_lb
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach1_wavelet.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach1_wavelet.py
index 64a88a2b63192fbb52940b88495eff46f2e54006..38a7312286d3fa90a640f71ed9e8a70cf050d138 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach1_wavelet.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach1_wavelet.py
@@ -1,14 +1,21 @@
"""
models/approach1_wavelet.py
Approach 1: Wavelet Decomposition CNN-LSTM
-With class weights to prevent majority-class collapse.
+- Dynamic wavelet level based on sequence length (no boundary warnings)
+- CPU-optimised smaller architecture
+- Class weights to prevent majority-class collapse
"""
import numpy as np
import pywt
WAVELET = "db4"
-LEVEL = 3
+
+
+def _safe_wavelet_level(lookback: int, wavelet: str = WAVELET) -> int:
+ """Compute max safe wavelet level for the given sequence length."""
+ max_level = pywt.dwt_max_level(lookback, wavelet)
+ return min(2, max_level) # cap at 2 to avoid boundary effects
def _wavelet_decompose_signal(signal: np.ndarray, wavelet: str, level: int) -> np.ndarray:
@@ -16,13 +23,14 @@ def _wavelet_decompose_signal(signal: np.ndarray, wavelet: str, level: int) -> n
coeffs = pywt.wavedec(signal, wavelet, level=level)
bands = []
for c in coeffs:
- band = np.interp(np.linspace(0, len(c)-1, T), np.arange(len(c)), c)
+ band = np.interp(np.linspace(0, len(c) - 1, T), np.arange(len(c)), c)
bands.append(band)
return np.stack(bands, axis=-1)
-def apply_wavelet_transform(X: np.ndarray, wavelet=WAVELET, level=LEVEL) -> np.ndarray:
+def apply_wavelet_transform(X: np.ndarray, wavelet: str = WAVELET) -> np.ndarray:
n_samples, lookback, n_features = X.shape
+ level = _safe_wavelet_level(lookback, wavelet)
n_bands = level + 1
X_wt = np.zeros((n_samples, lookback, n_features * n_bands), dtype=np.float32)
for s in range(n_samples):
@@ -33,18 +41,18 @@ def apply_wavelet_transform(X: np.ndarray, wavelet=WAVELET, level=LEVEL) -> np.n
return X_wt
-def build_wavelet_cnn_lstm(input_shape, n_classes, dropout=0.3, lstm_units=128):
+def build_wavelet_cnn_lstm(input_shape, n_classes, dropout=0.3, lstm_units=64):
from tensorflow import keras
from models.base import classification_head
inputs = keras.Input(shape=input_shape)
- x = keras.layers.Conv1D(64, 3, padding="causal", activation="relu")(inputs)
+ x = keras.layers.Conv1D(32, 3, padding="causal", activation="relu")(inputs)
x = keras.layers.BatchNormalization()(x)
x = keras.layers.MaxPooling1D(2)(x)
- x = keras.layers.Conv1D(32, 3, padding="causal", activation="relu")(x)
+ x = keras.layers.Conv1D(16, 3, padding="causal", activation="relu")(x)
x = keras.layers.BatchNormalization()(x)
x = keras.layers.Dropout(dropout)(x)
- x = keras.layers.LSTM(lstm_units, dropout=dropout, recurrent_dropout=0.1)(x)
+ x = keras.layers.LSTM(lstm_units, dropout=dropout)(x)
outputs = classification_head(x, n_classes, dropout)
model = keras.Model(inputs, outputs, name="Approach1_Wavelet")
@@ -58,7 +66,7 @@ def build_wavelet_cnn_lstm(input_shape, n_classes, dropout=0.3, lstm_units=128):
def train_approach1(
X_train, y_train, X_val, y_val,
- n_classes, epochs=100, batch_size=32, dropout=0.3, lstm_units=128,
+ n_classes, epochs=80, batch_size=64, dropout=0.3, lstm_units=64,
):
from models.base import get_callbacks, compute_class_weights
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach2_regime.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach2_regime.py
index 876b6d1f75c7b37a76b8464349043a3236df07be..f16c38ae40ed06d65e920cd6726fb92b5eeb96e9 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach2_regime.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach2_regime.py
@@ -1,29 +1,18 @@
"""
models/approach2_regime.py
Approach 2: Regime-Conditioned CNN-LSTM
-
-Pipeline:
- Raw macro signals
- -> CNN Tower (64 filters, k=3) -> feature vector
- -> Regime Classifier (HMM on VIX + HY spread + T10Y2Y) -> one-hot [4]
- -> Concatenate CNN features + regime embedding
- -> LSTM (128 units)
- -> Dense 64 -> Softmax (n_etfs + 1 CASH)
-
-NOTE: tensorflow and hmmlearn are imported lazily inside functions
-to prevent module-level import failures from making this module
-appear broken to Python's import system.
+- Fixed HMM convergence settings
+- CPU-optimised smaller architecture
+- Lazy imports to prevent module-level failures
+- Class weights to prevent majority-class collapse
"""
import numpy as np
-N_REGIMES = 4
-REGIME_HINTS = ["VIX", "HY", "Spread", "T10Y2Y", "T10Y3M", "Credit"]
-
+N_REGIMES = 3 # reduced from 4 to improve HMM convergence
+REGIME_HINTS = ["VIX", "HY", "Spread", "T10Y2Y", "T10Y3M", "Credit",
+ "IG_SPREAD", "HY_SPREAD"]
-# ---------------------------------------------------------------------------
-# Regime detection helpers
-# ---------------------------------------------------------------------------
def _get_regime_cols(feature_names: list) -> list:
return [
@@ -34,19 +23,12 @@ def _get_regime_cols(feature_names: list) -> list:
def fit_regime_model(X_flat: np.ndarray, feature_names: list,
n_regimes: int = N_REGIMES):
- """
- Fit a Gaussian HMM on regime-relevant macro features.
- Returns (hmm_model, regime_cols_idx).
- hmm_model is None if hmmlearn is unavailable or fitting fails.
- """
regime_col_names = _get_regime_cols(feature_names)
if not regime_col_names:
regime_col_names = feature_names[:min(3, len(feature_names))]
- regime_cols_idx = [
- feature_names.index(c) for c in regime_col_names
- if c in feature_names
- ]
+ regime_cols_idx = [feature_names.index(c) for c in regime_col_names
+ if c in feature_names]
X_regime = X_flat[:, regime_cols_idx]
try:
@@ -54,70 +36,55 @@ def fit_regime_model(X_flat: np.ndarray, feature_names: list,
hmm = GaussianHMM(
n_components=n_regimes,
covariance_type="diag",
- n_iter=100,
+ n_iter=50, # reduced from 100
+ tol=1e-2, # looser tolerance β avoids non-convergence warning
random_state=42,
)
hmm.fit(X_regime)
return hmm, regime_cols_idx
except Exception as e:
- print(f"[Approach 2] HMM fitting failed: {e}. Using fallback.")
+ print(f"[Approach 2] HMM failed: {e}. Using quantile fallback.")
return None, regime_cols_idx
def predict_regimes(hmm_model, X_flat: np.ndarray,
regime_cols_idx: list,
n_regimes: int = N_REGIMES) -> np.ndarray:
- """Predict integer regime label for each day."""
X_regime = X_flat[:, regime_cols_idx]
-
if hmm_model is not None:
try:
return hmm_model.predict(X_regime)
except Exception:
pass
-
- # Fallback: quantile binning on first regime feature
feat = X_regime[:, 0]
quantiles = np.percentile(feat, np.linspace(0, 100, n_regimes + 1))
return np.digitize(feat, quantiles[1:-1]).astype(int)
-def regimes_to_onehot(regimes: np.ndarray,
- n_regimes: int = N_REGIMES) -> np.ndarray:
+def regimes_to_onehot(regimes: np.ndarray, n_regimes: int = N_REGIMES) -> np.ndarray:
one_hot = np.zeros((len(regimes), n_regimes), dtype=np.float32)
for i, r in enumerate(regimes):
one_hot[i, min(int(r), n_regimes - 1)] = 1.0
return one_hot
-def build_regime_sequences(X_seq: np.ndarray,
- regimes_flat: np.ndarray,
+def build_regime_sequences(X_seq: np.ndarray, regimes_flat: np.ndarray,
lookback: int) -> np.ndarray:
n_samples = X_seq.shape[0]
aligned = regimes_flat[lookback: lookback + n_samples]
return regimes_to_onehot(aligned)
-# ---------------------------------------------------------------------------
-# Model builder
-# ---------------------------------------------------------------------------
-
-def build_regime_cnn_lstm(seq_input_shape: tuple,
- n_classes: int,
- n_regimes: int = N_REGIMES,
- dropout: float = 0.3,
- lstm_units: int = 128):
- """Build and compile the regime-conditioned CNN-LSTM model."""
+def build_regime_cnn_lstm(seq_input_shape, n_classes,
+ n_regimes=N_REGIMES, dropout=0.3, lstm_units=64):
from tensorflow import keras
from models.base import classification_head
seq_input = keras.Input(shape=seq_input_shape, name="seq_input")
- x = keras.layers.Conv1D(64, kernel_size=3, padding="causal",
- activation="relu")(seq_input)
+ x = keras.layers.Conv1D(32, 3, padding="causal", activation="relu")(seq_input)
x = keras.layers.BatchNormalization()(x)
- x = keras.layers.MaxPooling1D(pool_size=2)(x)
- x = keras.layers.Conv1D(32, kernel_size=3, padding="causal",
- activation="relu")(x)
+ x = keras.layers.MaxPooling1D(2)(x)
+ x = keras.layers.Conv1D(16, 3, padding="causal", activation="relu")(x)
x = keras.layers.BatchNormalization()(x)
x = keras.layers.Dropout(dropout)(x)
cnn_out = keras.layers.GlobalAveragePooling1D()(x)
@@ -128,59 +95,35 @@ def build_regime_cnn_lstm(seq_input_shape: tuple,
merged = keras.layers.Concatenate()([cnn_out, regime_emb])
x = keras.layers.Reshape((1, merged.shape[-1]))(merged)
x = keras.layers.LSTM(lstm_units, dropout=dropout)(x)
-
outputs = classification_head(x, n_classes, dropout)
- model = keras.Model(
- inputs=[seq_input, regime_input],
- outputs=outputs,
- name="Approach2_Regime_CNN_LSTM",
- )
+ model = keras.Model(inputs=[seq_input, regime_input], outputs=outputs,
+ name="Approach2_Regime")
model.compile(
- optimizer=keras.optimizers.Adam(learning_rate=1e-3),
+ optimizer=keras.optimizers.Adam(1e-3),
loss="sparse_categorical_crossentropy",
metrics=["accuracy"],
)
return model
-# ---------------------------------------------------------------------------
-# Training pipeline
-# ---------------------------------------------------------------------------
-
def train_approach2(
- X_train, y_train,
- X_val, y_val,
- X_flat_all: np.ndarray,
- feature_names: list,
- lookback: int,
- train_size: int,
- val_size: int,
- n_classes: int,
- epochs: int = 100,
- batch_size: int = 32,
- dropout: float = 0.3,
- lstm_units: int = 128,
+ X_train, y_train, X_val, y_val,
+ X_flat_all, feature_names, lookback,
+ train_size, val_size, n_classes,
+ epochs=80, batch_size=64, dropout=0.3, lstm_units=64,
):
- """
- Fit HMM regime model then train the regime-conditioned CNN-LSTM.
- Returns: model, history, hmm_model, regime_cols_idx
- """
from models.base import get_callbacks, compute_class_weights
X_flat_train = X_flat_all[:train_size + lookback]
hmm_model, regime_cols_idx = fit_regime_model(X_flat_train, feature_names)
-
regimes_all = predict_regimes(hmm_model, X_flat_all, regime_cols_idx)
R_train = build_regime_sequences(X_train, regimes_all, lookback)
R_val = build_regime_sequences(X_val, regimes_all, lookback + train_size)
- model = build_regime_cnn_lstm(
- X_train.shape[1:], n_classes,
- dropout=dropout, lstm_units=lstm_units,
- )
-
+ model = build_regime_cnn_lstm(X_train.shape[1:], n_classes,
+ dropout=dropout, lstm_units=lstm_units)
cw = compute_class_weights(y_train, n_classes)
history = model.fit(
@@ -192,29 +135,13 @@ def train_approach2(
callbacks=get_callbacks(),
verbose=0,
)
-
return model, history, hmm_model, regime_cols_idx
-# ---------------------------------------------------------------------------
-# Inference
-# ---------------------------------------------------------------------------
-
-def predict_approach2(
- model,
- X_test: np.ndarray,
- X_flat_all: np.ndarray,
- regime_cols_idx: list,
- hmm_model,
- lookback: int,
- train_size: int,
- val_size: int,
-) -> tuple:
- """Predict on test set with regime conditioning. Returns (preds, proba)."""
+def predict_approach2(model, X_test, X_flat_all, regime_cols_idx,
+ hmm_model, lookback, train_size, val_size) -> tuple:
regimes_all = predict_regimes(hmm_model, X_flat_all, regime_cols_idx)
offset = lookback + train_size + val_size
R_test = build_regime_sequences(X_test, regimes_all, offset)
-
- proba = model.predict([X_test, R_test], verbose=0)
- preds = np.argmax(proba, axis=1)
- return preds, proba
+ proba = model.predict([X_test, R_test], verbose=0)
+ return np.argmax(proba, axis=1), proba
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach3_multiscale.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach3_multiscale.py
index 09014aae1de9efef3f00618cfdac27addc7aed4c..d1983181b046f8bb05c7757a65deab72f18e43fc 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach3_multiscale.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/approach3_multiscale.py
@@ -1,18 +1,21 @@
"""
models/approach3_multiscale.py
Approach 3: Multi-Scale Parallel CNN-LSTM
-With class weights to prevent majority-class collapse.
+- CPU-optimised smaller architecture
+- Class weights to prevent majority-class collapse
+- Lazy imports to prevent module-level failures
"""
import numpy as np
KERNEL_SIZES = [3, 7, 21]
-FILTERS_EACH = 32
+FILTERS_EACH = 16 # reduced from 32 for CPU speed
def build_multiscale_cnn_lstm(
- input_shape, n_classes, kernel_sizes=None,
- filters=FILTERS_EACH, dropout=0.3, lstm_units=128,
+ input_shape, n_classes,
+ kernel_sizes=None, filters=FILTERS_EACH,
+ dropout=0.3, lstm_units=64,
):
from tensorflow import keras
from models.base import classification_head
@@ -22,19 +25,15 @@ def build_multiscale_cnn_lstm(
inputs = keras.Input(shape=input_shape, name="multiscale_input")
towers = []
-
for k in kernel_sizes:
t = keras.layers.Conv1D(filters, k, padding="causal", activation="relu",
name=f"conv1_k{k}")(inputs)
t = keras.layers.BatchNormalization(name=f"bn1_k{k}")(t)
- t = keras.layers.Conv1D(filters, k, padding="causal", activation="relu",
- name=f"conv2_k{k}")(t)
- t = keras.layers.BatchNormalization(name=f"bn2_k{k}")(t)
t = keras.layers.Dropout(dropout, name=f"drop_k{k}")(t)
towers.append(t)
- merged = keras.layers.Concatenate(axis=-1)(towers) if len(towers) > 1 else towers[0]
- x = keras.layers.LSTM(lstm_units, dropout=dropout, recurrent_dropout=0.1)(merged)
+ merged = keras.layers.Concatenate(axis=-1)(towers) if len(towers) > 1 else towers[0]
+ x = keras.layers.LSTM(lstm_units, dropout=dropout)(merged)
outputs = classification_head(x, n_classes, dropout)
model = keras.Model(inputs, outputs, name="Approach3_MultiScale")
@@ -48,8 +47,8 @@ def build_multiscale_cnn_lstm(
def train_approach3(
X_train, y_train, X_val, y_val,
- n_classes, epochs=100, batch_size=32,
- dropout=0.3, lstm_units=128, kernel_sizes=None,
+ n_classes, epochs=80, batch_size=64,
+ dropout=0.3, lstm_units=64, kernel_sizes=None,
):
from models.base import get_callbacks, compute_class_weights
@@ -59,7 +58,8 @@ def train_approach3(
lookback = X_train.shape[1]
valid_kernels = [k for k in kernel_sizes if k <= lookback] or [min(3, lookback)]
model = build_multiscale_cnn_lstm(
- X_train.shape[1:], n_classes, valid_kernels, dropout=dropout, lstm_units=lstm_units,
+ X_train.shape[1:], n_classes, valid_kernels,
+ dropout=dropout, lstm_units=lstm_units,
)
cw = compute_class_weights(y_train, n_classes)
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
index b020ba27880c8a58a313304eab320de88361dd57..9d6c137e9000054c5531972594c2d085e74782b0 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
@@ -1,200 +1,171 @@
"""
-models/base.py
-Shared utilities for all CNN-LSTM variants.
-Optimised for CPU training on HF Spaces.
+strategy/backtest.py
+Strategy execution, performance metrics, and benchmark calculations.
+
+CASH logic (drawdown risk overlay β not a model class):
+ ENTER : 2-day cumulative return <= -15%
+ EXIT : model conviction Z-score >= 1.0 (model decisively picks an ETF again)
"""
import numpy as np
-import hashlib
-import pickle
-import os
-from pathlib import Path
-from sklearn.preprocessing import RobustScaler
-from sklearn.utils.class_weight import compute_class_weight
-
-SEED = 42
-CACHE_DIR = Path("/tmp/p2_model_cache")
-CACHE_DIR.mkdir(exist_ok=True)
-
-np.random.seed(SEED)
-
-
-# ββ Cache helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def make_cache_key(last_date: str, start_yr: int, fee_bps: int,
- epochs: int, split: str, include_cash: bool,
- lookback: int) -> str:
- raw = f"{last_date}_{start_yr}_{fee_bps}_{epochs}_{split}_{include_cash}_{lookback}"
- return hashlib.md5(raw.encode()).hexdigest()
-
-
-def save_cache(key: str, payload: dict):
- path = CACHE_DIR / f"{key}.pkl"
- with open(path, "wb") as f:
- pickle.dump(payload, f)
-
-
-def load_cache(key: str) -> dict | None:
- path = CACHE_DIR / f"{key}.pkl"
- if path.exists():
- try:
- with open(path, "rb") as f:
- return pickle.load(f)
- except Exception:
- path.unlink(missing_ok=True)
- return None
-
-
-# ββ Sequence builder ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def build_sequences(features: np.ndarray, targets: np.ndarray, lookback: int):
- X, y = [], []
- for i in range(lookback, len(features)):
- X.append(features[i - lookback: i])
- y.append(targets[i])
- return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32)
-
-
-# ββ Train / val / test split ββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def train_val_test_split(X, y, train_pct=0.70, val_pct=0.15):
- n = len(X)
- t1 = int(n * train_pct)
- t2 = int(n * (train_pct + val_pct))
- return X[:t1], y[:t1], X[t1:t2], y[t1:t2], X[t2:], y[t2:]
-
-
-# ββ Feature scaling βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def scale_features(X_train, X_val, X_test):
- n_feat = X_train.shape[2]
- scaler = RobustScaler()
- scaler.fit(X_train.reshape(-1, n_feat))
- def _t(X):
- s = X.shape
- return scaler.transform(X.reshape(-1, n_feat)).reshape(s)
- return _t(X_train), _t(X_val), _t(X_test), scaler
-
-
-# ββ Label builder βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def returns_to_labels(y_raw, include_cash=True, cash_threshold=0.0):
- best = np.argmax(y_raw, axis=1)
- if include_cash:
- best_ret = y_raw[np.arange(len(y_raw)), best]
- cash_idx = y_raw.shape[1]
- labels = np.where(best_ret < cash_threshold, cash_idx, best)
- else:
- labels = best
- return labels.astype(np.int32)
-
-
-# ββ Class weights βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def compute_class_weights(y_labels: np.ndarray, n_classes: int) -> dict:
- present = np.unique(y_labels)
- try:
- weights = compute_class_weight("balanced", classes=present, y=y_labels)
- weight_dict = {int(c): float(w) for c, w in zip(present, weights)}
- except Exception:
- weight_dict = {}
- for c in range(n_classes):
- if c not in weight_dict:
- weight_dict[c] = 1.0
- return weight_dict
-
-
-# ββ Callbacks βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def get_callbacks(patience_es=15, patience_lr=8, min_lr=1e-6):
- from tensorflow import keras
- return [
- keras.callbacks.EarlyStopping(
- monitor="val_loss",
- patience=patience_es,
- restore_best_weights=True,
- verbose=0,
- ),
- keras.callbacks.ReduceLROnPlateau(
- monitor="val_loss",
- factor=0.5,
- patience=patience_lr,
- min_lr=min_lr,
- verbose=0,
- ),
- ]
-
-
-# ββ Lightweight output head (CPU-optimised) βββββββββββββββββββββββββββββββββββ
-
-def classification_head(x, n_classes: int, dropout: float = 0.3):
- """Smaller head than original β faster on CPU, less overfitting risk."""
- from tensorflow import keras
- x = keras.layers.Dense(32, activation="relu")(x)
- x = keras.layers.Dropout(dropout)(x)
- x = keras.layers.Dense(n_classes, activation="softmax")(x)
- return x
-
-
-# ββ Auto lookback selection βββββββββββββββββββββββββββββββββββββββββββββββββββ
-
-def find_best_lookback(
- X_raw: np.ndarray,
- y_raw: np.ndarray,
- y_labels_fn,
- train_pct: float,
- val_pct: float,
- n_classes: int,
- include_cash: bool,
- candidates: list = None,
-):
- """
- Train a fast lightweight CNN on each lookback candidate using val loss.
- Returns best lookback int.
- Uses only Approach 1 architecture (fastest) to pick the winner.
- """
- from tensorflow import keras
-
- if candidates is None:
- candidates = [30, 45, 60]
-
- best_lb = candidates[0]
- best_loss = np.inf
-
- for lb in candidates:
- try:
- X_seq, y_seq = build_sequences(X_raw, y_raw, lb)
- y_lab = y_labels_fn(y_seq)
-
- X_tr, y_tr, X_v, y_v, _, _ = train_val_test_split(X_seq, y_lab, train_pct, val_pct)
- X_tr_s, X_v_s, _, _ = scale_features(X_tr, X_v, X_v)
-
- cw = compute_class_weights(y_tr, n_classes)
-
- # Tiny fast model just for lookback selection
- inp = keras.Input(shape=X_tr_s.shape[1:])
- x = keras.layers.Conv1D(16, min(3, lb), padding="causal", activation="relu")(inp)
- x = keras.layers.GlobalAveragePooling1D()(x)
- out = keras.layers.Dense(n_classes, activation="softmax")(x)
- m = keras.Model(inp, out)
- m.compile(optimizer="adam", loss="sparse_categorical_crossentropy")
-
- hist = m.fit(
- X_tr_s, y_tr,
- validation_data=(X_v_s, y_v),
- epochs=15,
- batch_size=64,
- class_weight=cw,
- callbacks=[keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True)],
- verbose=0,
- )
- val_loss = min(hist.history.get("val_loss", [np.inf]))
- if val_loss < best_loss:
- best_loss = val_loss
- best_lb = lb
-
- del m
- except Exception:
+import pandas as pd
+from datetime import datetime
+
+CASH_DRAWDOWN_TRIGGER = -0.15 # 2-day cumulative return threshold
+CASH_EXIT_Z = 1.0 # Z-score required to exit CASH
+
+
+def _zscore(proba: np.ndarray) -> float:
+ std = np.std(proba)
+ return float((np.max(proba) - np.mean(proba)) / std) if std > 1e-9 else 0.0
+
+
+def execute_strategy(
+ preds: np.ndarray,
+ proba: np.ndarray,
+ y_raw_test: np.ndarray,
+ test_dates: pd.DatetimeIndex,
+ target_etfs: list,
+ fee_bps: int,
+ tbill_rate: float,
+ include_cash: bool = True, # kept for API compat but CASH is now overlay-only
+) -> dict:
+ n_etfs = len(target_etfs)
+ daily_tbill = tbill_rate / 252
+ fee = fee_bps / 10000
+ today = datetime.now().date()
+
+ strat_rets = []
+ audit_trail = []
+ date_index = []
+
+ in_cash = False
+ recent_rets = [] # rolling 2-day window
+
+ for i, cls in enumerate(preds):
+ cls = min(int(cls), n_etfs - 1)
+ etf_name = target_etfs[cls].replace("_Ret", "")
+ etf_ret = float(np.clip(y_raw_test[i][cls], -0.5, 0.5))
+ z = _zscore(proba[i])
+
+ # ββ 2-day drawdown check ββββββββββββββββββββββββββββββββββββββββββββββ
+ recent_rets.append(etf_ret)
+ if len(recent_rets) > 2:
+ recent_rets.pop(0)
+ two_day = ((1 + recent_rets[0]) * (1 + recent_rets[-1]) - 1
+ if len(recent_rets) >= 2 else 0.0)
+
+ if two_day <= CASH_DRAWDOWN_TRIGGER:
+ in_cash = True
+ if in_cash and z >= CASH_EXIT_Z:
+ in_cash = False
+
+ # ββ Execute βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+ if in_cash:
+ signal_etf = "CASH"
+ realized_ret = daily_tbill
+ else:
+ signal_etf = etf_name
+ realized_ret = etf_ret
+
+ net_ret = realized_ret - fee
+ strat_rets.append(net_ret)
+ date_index.append(test_dates[i])
+
+ if test_dates[i].date() < today:
+ audit_trail.append({
+ "Date": test_dates[i].strftime("%Y-%m-%d"),
+ "Signal": signal_etf,
+ "Net_Return": net_ret,
+ "Z_Score": round(z, 2),
+ })
+
+ strat_rets = np.array(strat_rets, dtype=np.float64)
+
+ # Next signal
+ last_cls = min(int(preds[-1]), n_etfs - 1)
+ last_z = _zscore(proba[-1])
+ last_ret = float(np.clip(y_raw_test[-1][last_cls], -0.5, 0.5))
+ prev_ret = float(np.clip(y_raw_test[-2][last_cls], -0.5, 0.5)) if len(y_raw_test) > 1 else 0.0
+ last_2d = (1 + prev_ret) * (1 + last_ret) - 1
+ next_cash = last_2d <= CASH_DRAWDOWN_TRIGGER and last_z < CASH_EXIT_Z
+ next_signal = "CASH" if next_cash else target_etfs[last_cls].replace("_Ret", "")
+
+ metrics = _compute_metrics(strat_rets, tbill_rate, date_index)
+
+ return {
+ **metrics,
+ "strat_rets": strat_rets,
+ "audit_trail": audit_trail,
+ "next_signal": next_signal,
+ "next_proba": proba[-1],
+ }
+
+
+def _compute_metrics(strat_rets: np.ndarray, tbill_rate: float,
+ date_index: list = None) -> dict:
+ if len(strat_rets) == 0:
+ return {}
+
+ cum_returns = np.cumprod(1 + strat_rets)
+ n = len(strat_rets)
+ ann_return = float(cum_returns[-1] ** (252 / n) - 1)
+
+ excess = strat_rets - tbill_rate / 252
+ sharpe = float(np.mean(excess) / (np.std(strat_rets) + 1e-9) * np.sqrt(252))
+
+ hit_ratio = float(np.mean(strat_rets[-15:] > 0))
+
+ cum_max = np.maximum.accumulate(cum_returns)
+ drawdown = (cum_returns - cum_max) / cum_max
+ max_dd = float(np.min(drawdown))
+
+ worst_idx = int(np.argmin(strat_rets))
+ max_daily = float(strat_rets[worst_idx])
+ worst_date = (date_index[worst_idx].strftime("%Y-%m-%d")
+ if date_index and worst_idx < len(date_index) else "N/A")
+
+ return {
+ "cum_returns": cum_returns,
+ "ann_return": ann_return,
+ "sharpe": sharpe,
+ "hit_ratio": hit_ratio,
+ "max_dd": max_dd,
+ "max_daily_dd": max_daily,
+ "max_daily_date": worst_date,
+ "cum_max": cum_max,
+ }
+
+
+def compute_benchmark_metrics(returns: np.ndarray, tbill_rate: float) -> dict:
+ return _compute_metrics(np.array(returns, dtype=np.float64), tbill_rate)
+
+
+def select_winner(results: dict) -> str:
+ best_name, best_ret = None, -np.inf
+ for name, res in results.items():
+ if res is None:
continue
-
- return best_lb
+ r = res.get("ann_return", -np.inf)
+ if r > best_ret:
+ best_ret, best_name = r, name
+ return best_name
+
+
+def build_comparison_table(results: dict, winner_name: str) -> pd.DataFrame:
+ rows = []
+ for name, res in results.items():
+ if res is None:
+ rows.append({"Approach": name, "Ann. Return": "N/A",
+ "Sharpe": "N/A", "Hit Ratio (15d)": "N/A",
+ "Max Drawdown": "N/A", "Winner": ""})
+ continue
+ rows.append({
+ "Approach": name,
+ "Ann. Return": f"{res['ann_return']*100:.2f}%",
+ "Sharpe": f"{res['sharpe']:.2f}",
+ "Hit Ratio (15d)": f"{res['hit_ratio']*100:.0f}%",
+ "Max Drawdown": f"{res['max_dd']*100:.2f}%",
+ "Winner": "β WINNER" if name == winner_name else "",
+ })
+ return pd.DataFrame(rows)
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py
index 48afd9682daacbe9ce9d72e86fd2c0cc5c8292cf..5f062a870f25bb5e93ede2fbc2d52de6aff150c0 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py
@@ -1,9 +1,11 @@
"""
ui/components.py
Reusable Streamlit UI blocks.
-- Fixed applymap β map deprecation
-- Removed debug expanders
-- Added show_all_signals_panel
+Changes:
+- Metrics row: Ann Return compared vs SPY (not T-bill)
+- Max Daily DD: shows date it happened
+- Conviction panel: compact single-line ETF list (no big bars)
+- applymap β map (deprecation fix)
"""
import streamlit as st
@@ -25,8 +27,10 @@ def show_freshness_status(freshness: dict):
def show_signal_banner(next_signal: str, next_date, approach_name: str):
is_cash = next_signal == "CASH"
- bg = ("linear-gradient(135deg, #2d3436 0%, #1a1a2e 100%)" if is_cash
- else "linear-gradient(135deg, #00d1b2 0%, #00a896 100%)")
+ bg = ("linear-gradient(135deg, #2d3436 0%, #1a1a2e 100%)" if is_cash
+ else "linear-gradient(135deg, #00d1b2 0%, #00a896 100%)")
+ label = ("β οΈ DRAWDOWN PROTECTION ACTIVE β CASH"
+ if is_cash else f"π― {next_date.strftime('%Y-%m-%d')} β {next_signal}")
st.markdown(f"""
@@ -34,9 +38,9 @@ def show_signal_banner(next_signal: str, next_date, approach_name: str):
letter-spacing:3px; margin-bottom:6px;">
{approach_name.upper()} Β· NEXT TRADING DAY SIGNAL
-
- π― {next_date.strftime('%Y-%m-%d')} β {next_signal}
+
+ {label}
""", unsafe_allow_html=True)
@@ -46,34 +50,25 @@ def show_signal_banner(next_signal: str, next_date, approach_name: str):
def show_all_signals_panel(all_signals: dict, target_etfs: list,
include_cash: bool, next_date, optimal_lookback: int):
- APPROACH_COLORS = {
- "Approach 1": "#00ffc8",
- "Approach 2": "#7c6aff",
- "Approach 3": "#ff6b6b",
- }
+ COLORS = {"Approach 1": "#00ffc8", "Approach 2": "#7c6aff", "Approach 3": "#ff6b6b"}
st.subheader(f"ποΈ All Models β {next_date.strftime('%Y-%m-%d')} Signals")
- st.caption(f"π Optimal lookback: **{optimal_lookback}d** (auto-selected from 30/45/60)")
+ st.caption(f"π Lookback **{optimal_lookback}d** found optimal (auto-selected from 30 / 45 / 60d)")
cols = st.columns(len(all_signals))
for col, (name, info) in zip(cols, all_signals.items()):
- color = APPROACH_COLORS.get(name, "#888888")
- signal = info["signal"]
- proba = info["proba"]
- top_prob = float(np.max(proba)) * 100
- is_winner = info["is_winner"]
- badge = " β" if is_winner else ""
+ color = COLORS.get(name, "#888")
+ signal = info["signal"]
+ top_prob = float(np.max(info["proba"])) * 100
+ badge = " β" if info["is_winner"] else ""
+ sig_col = "#aaa" if signal == "CASH" else "white"
col.markdown(f"""
@@ -93,14 +88,12 @@ def show_conviction_panel(conviction: dict):
z_clipped = max(-3.0, min(3.0, z_score))
bar_pct = int((z_clipped + 3) / 6 * 100)
- max_score = max((s for _, s in sorted_pairs), default=1.0)
- if max_score <= 0:
- max_score = 1.0
+ # ββ Header with Z-score gauge βββββββββββββββββββββββββββββββββββββββββββββ
st.markdown(f"""
{icon}
@@ -118,76 +111,73 @@ def show_conviction_panel(conviction: dict):
font-size:11px; color:#999; margin-bottom:4px;">
Weak −3σNeutral 0σStrong +3σ
-
+
-
- Model probability by ETF (ranked high → low):
+
+ MODEL PROBABILITY BY ETF
+
+
+ {"".join([
+ f''
+ f'{"β
" if n == best_name else ""}{n} {s:.3f}'
+ for n, s in sorted_pairs
+ ])}
""", unsafe_allow_html=True)
- for i, (name, score) in enumerate(sorted_pairs):
- is_winner = (name == best_name)
- is_last = (i == len(sorted_pairs) - 1)
- bar_w = int(score / max_score * 100)
- name_style = "font-weight:700; color:#00897b;" if is_winner else "color:#444;"
- bar_color = color if is_winner else "#b2dfdb" if score > max_score * 0.5 else "#e0e0e0"
- star = " β
" if is_winner else ""
- bottom_r = "0 0 12px 12px" if is_last else "0"
- border_bot = "border-bottom:1px solid #f0f0f0;" if not is_last else ""
-
- st.markdown(f"""
-
-
-
{name}{star}
-
-
{score:.4f}
-
-
- """, unsafe_allow_html=True)
-
st.caption(
- "Z-score = std deviations the top ETF's probability sits above the mean. "
- "Higher β model is more decisive."
+ "Z-score = std deviations the top ETF's probability sits above the mean of all ETF probabilities. "
+ "Higher β model is more decisive. "
+ "β οΈ CASH override triggers if 2-day cumulative return β€ β15%, exits when Z β₯ 1.0."
)
# ββ Metrics row βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
-def show_metrics_row(result: dict, tbill_rate: float):
- col1, col2, col3, col4, col5 = st.columns(5)
- col1.metric("π Ann. Return", f"{result['ann_return']*100:.2f}%",
- delta=f"vs T-bill: {(result['ann_return'] - tbill_rate)*100:.2f}%")
- col2.metric("π Sharpe", f"{result['sharpe']:.2f}",
- delta="Strong" if result['sharpe'] > 1 else "Weak")
- col3.metric("π― Hit Ratio 15d", f"{result['hit_ratio']*100:.0f}%",
- delta="Good" if result['hit_ratio'] > 0.55 else "Weak")
- col4.metric("π Max Drawdown", f"{result['max_dd']*100:.2f}%",
- delta="Peak to Trough")
- col5.metric("β οΈ Max Daily DD", f"{result['max_daily_dd']*100:.2f}%",
- delta="Worst Day")
+def show_metrics_row(result: dict, tbill_rate: float, spy_ann_return: float = None):
+ c1, c2, c3, c4, c5 = st.columns(5)
+
+ # Ann return vs SPY
+ if spy_ann_return is not None:
+ diff = (result['ann_return'] - spy_ann_return) * 100
+ sign = "+" if diff >= 0 else ""
+ delta_str = f"vs SPY: {sign}{diff:.2f}%"
+ else:
+ delta_str = f"vs T-bill: {(result['ann_return'] - tbill_rate)*100:.2f}%"
+
+ c1.metric("π Ann. Return", f"{result['ann_return']*100:.2f}%", delta=delta_str)
+ c2.metric("π Sharpe", f"{result['sharpe']:.2f}",
+ delta="Strong" if result['sharpe'] > 1 else "Weak")
+ c3.metric("π― Hit Ratio 15d", f"{result['hit_ratio']*100:.0f}%",
+ delta="Good" if result['hit_ratio'] > 0.55 else "Weak")
+ c4.metric("π Max Drawdown", f"{result['max_dd']*100:.2f}%",
+ delta="Peak to Trough")
+
+ # Max daily DD with date
+ worst_date = result.get("max_daily_date", "N/A")
+ c5.metric("β οΈ Max Daily DD", f"{result['max_daily_dd']*100:.2f}%",
+ delta=f"on {worst_date}")
# ββ Comparison table ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def show_comparison_table(comparison_df: pd.DataFrame):
- def highlight_winner(row):
+ def _highlight(row):
if "WINNER" in str(row.get("Winner", "")):
return ["background-color: rgba(0,200,150,0.15); font-weight:bold"] * len(row)
return [""] * len(row)
styled = (
comparison_df.style
- .apply(highlight_winner, axis=1)
+ .apply(_highlight, axis=1)
.set_properties(**{"text-align": "center", "font-size": "14px"})
.set_table_styles([
{"selector": "th", "props": [("font-size", "14px"),
@@ -206,16 +196,22 @@ def show_audit_trail(audit_trail: list):
st.info("No audit trail data available.")
return
- df = pd.DataFrame(audit_trail).tail(20)[["Date", "Signal", "Net_Return"]]
+ df = pd.DataFrame(audit_trail).tail(20)
+ cols = [c for c in ["Date", "Signal", "Net_Return", "Z_Score"] if c in df.columns]
+ df = df[cols]
- def color_return(val):
+ def _color_ret(val):
return ("color: #00c896; font-weight:bold" if val > 0
else "color: #ff4b4b; font-weight:bold")
+ fmt = {"Net_Return": "{:.2%}"}
+ if "Z_Score" in df.columns:
+ fmt["Z_Score"] = "{:.2f}"
+
styled = (
df.style
- .map(color_return, subset=["Net_Return"])
- .format({"Net_Return": "{:.2%}"})
+ .map(_color_ret, subset=["Net_Return"])
+ .format(fmt)
.set_properties(**{"font-size": "14px", "text-align": "center"})
.set_table_styles([
{"selector": "th", "props": [("font-size", "14px"),
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py
index 3965d4d7256f2fd352e4ec58bb1b30bfb2c3fb5c..13e3961f6d6ae4013bbb4a26da3202927580491d 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/charts.py
@@ -1,14 +1,15 @@
"""
ui/charts.py
-Plotly chart builders.
-Equity curve: winner + SPY + AGG only. Y-axis as % growth (not raw multiplier).
+Equity curve: winner vs SPY and AGG only.
+Y-axis: % cumulative growth from 0 (not raw multiplier).
+SPY/AGG returns are verified as pct returns (clipped) before compounding.
"""
import numpy as np
import pandas as pd
import plotly.graph_objects as go
-WINNER_COLOUR = "#00ffc8"
+WINNER_COLOUR = "#00ffc8"
BENCHMARK_COLOURS = {"SPY": "#ff4b4b", "AGG": "#ffa500"}
@@ -20,10 +21,6 @@ def equity_curve_chart(
test_slice: slice,
tbill_rate: float,
) -> go.Figure:
- """
- Equity curve: winner strategy vs SPY and AGG.
- Y-axis shows % growth (cum_return - 1) * 100 for readability.
- """
from strategy.backtest import compute_benchmark_metrics
fig = go.Figure()
@@ -32,46 +29,55 @@ def equity_curve_chart(
winner_res = results.get(winner_name)
if winner_res is not None:
cum = winner_res["cum_returns"]
- n = min(len(cum), len(plot_dates))
- fig.add_trace(go.Scatter(
- x=plot_dates[:n],
- y=(cum[:n] - 1) * 100,
- mode="lines",
- name=f"{winner_name} β
",
- line=dict(color=WINNER_COLOUR, width=2.5),
- fill="tozeroy",
- fillcolor="rgba(0,255,200,0.07)",
- ))
+ # Sanity: if cum[-1] > 10x (1000%), something is wrong β skip render
+ if cum[-1] < 10:
+ n = min(len(cum), len(plot_dates))
+ pct = (cum[:n] - 1) * 100
+ fig.add_trace(go.Scatter(
+ x=plot_dates[:n], y=pct,
+ mode="lines",
+ name=f"{winner_name} β
",
+ line=dict(color=WINNER_COLOUR, width=2.5),
+ fill="tozeroy",
+ fillcolor="rgba(0,255,200,0.07)",
+ ))
- # ββ SPY benchmark βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+ # ββ SPY βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+ spy_ann = None
if "SPY_Ret" in df.columns:
- spy_rets = df["SPY_Ret"].iloc[test_slice].values.copy()
- spy_rets = np.clip(spy_rets, -0.5, 0.5) # sanity clip
- spy_rets = spy_rets[~np.isnan(spy_rets)]
- n = min(len(spy_rets), len(plot_dates))
- spy_m = compute_benchmark_metrics(spy_rets[:n], tbill_rate)
- fig.add_trace(go.Scatter(
- x=plot_dates[:n],
- y=(spy_m["cum_returns"] - 1) * 100,
- mode="lines",
- name="SPY",
- line=dict(color=BENCHMARK_COLOURS["SPY"], width=1.5, dash="dot"),
- ))
+ raw = df["SPY_Ret"].iloc[test_slice].values.copy().astype(float)
+ raw = raw[~np.isnan(raw)]
+ # If values look like prices (median > 1), convert to returns
+ if len(raw) > 0 and np.median(np.abs(raw)) > 1:
+ raw = np.diff(raw) / raw[:-1]
+ raw = np.clip(raw, -0.5, 0.5)
+ if len(raw) > 0:
+ n = min(len(raw), len(plot_dates))
+ spy_m = compute_benchmark_metrics(raw[:n], tbill_rate)
+ spy_ann = spy_m.get("ann_return")
+ fig.add_trace(go.Scatter(
+ x=plot_dates[:n],
+ y=(spy_m["cum_returns"] - 1) * 100,
+ mode="lines", name="SPY",
+ line=dict(color=BENCHMARK_COLOURS["SPY"], width=1.5, dash="dot"),
+ ))
- # ββ AGG benchmark βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+ # ββ AGG βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
if "AGG_Ret" in df.columns:
- agg_rets = df["AGG_Ret"].iloc[test_slice].values.copy()
- agg_rets = np.clip(agg_rets, -0.5, 0.5)
- agg_rets = agg_rets[~np.isnan(agg_rets)]
- n = min(len(agg_rets), len(plot_dates))
- agg_m = compute_benchmark_metrics(agg_rets[:n], tbill_rate)
- fig.add_trace(go.Scatter(
- x=plot_dates[:n],
- y=(agg_m["cum_returns"] - 1) * 100,
- mode="lines",
- name="AGG",
- line=dict(color=BENCHMARK_COLOURS["AGG"], width=1.5, dash="dot"),
- ))
+ raw = df["AGG_Ret"].iloc[test_slice].values.copy().astype(float)
+ raw = raw[~np.isnan(raw)]
+ if len(raw) > 0 and np.median(np.abs(raw)) > 1:
+ raw = np.diff(raw) / raw[:-1]
+ raw = np.clip(raw, -0.5, 0.5)
+ if len(raw) > 0:
+ n = min(len(raw), len(plot_dates))
+ agg_m = compute_benchmark_metrics(raw[:n], tbill_rate)
+ fig.add_trace(go.Scatter(
+ x=plot_dates[:n],
+ y=(agg_m["cum_returns"] - 1) * 100,
+ mode="lines", name="AGG",
+ line=dict(color=BENCHMARK_COLOURS["AGG"], width=1.5, dash="dot"),
+ ))
fig.update_layout(
template="plotly_dark",
@@ -80,13 +86,7 @@ def equity_curve_chart(
legend=dict(yanchor="top", y=0.99, xanchor="left", x=0.01, font=dict(size=11)),
xaxis_title="Date",
yaxis_title="Cumulative Return (%)",
- margin=dict(l=50, r=30, t=20, b=50),
yaxis=dict(ticksuffix="%"),
+ margin=dict(l=50, r=30, t=20, b=50),
)
- return fig
-
-
-def _hex_to_rgb(hex_color: str) -> str:
- h = hex_color.lstrip("#")
- r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
- return f"{r},{g},{b}"
+ return fig, spy_ann
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
index 9d6c137e9000054c5531972594c2d085e74782b0..36fe25e40e923799b2a57e008d8aba1d83178e3d 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
@@ -1,171 +1,162 @@
"""
-strategy/backtest.py
-Strategy execution, performance metrics, and benchmark calculations.
-
-CASH logic (drawdown risk overlay β not a model class):
- ENTER : 2-day cumulative return <= -15%
- EXIT : model conviction Z-score >= 1.0 (model decisively picks an ETF again)
+models/base.py
+Shared utilities for all CNN-LSTM variants.
+Optimised for CPU training on HF Spaces.
"""
import numpy as np
-import pandas as pd
-from datetime import datetime
-
-CASH_DRAWDOWN_TRIGGER = -0.15 # 2-day cumulative return threshold
-CASH_EXIT_Z = 1.0 # Z-score required to exit CASH
-
-
-def _zscore(proba: np.ndarray) -> float:
- std = np.std(proba)
- return float((np.max(proba) - np.mean(proba)) / std) if std > 1e-9 else 0.0
-
-
-def execute_strategy(
- preds: np.ndarray,
- proba: np.ndarray,
- y_raw_test: np.ndarray,
- test_dates: pd.DatetimeIndex,
- target_etfs: list,
- fee_bps: int,
- tbill_rate: float,
- include_cash: bool = True, # kept for API compat but CASH is now overlay-only
-) -> dict:
- n_etfs = len(target_etfs)
- daily_tbill = tbill_rate / 252
- fee = fee_bps / 10000
- today = datetime.now().date()
-
- strat_rets = []
- audit_trail = []
- date_index = []
-
- in_cash = False
- recent_rets = [] # rolling 2-day window
-
- for i, cls in enumerate(preds):
- cls = min(int(cls), n_etfs - 1)
- etf_name = target_etfs[cls].replace("_Ret", "")
- etf_ret = float(np.clip(y_raw_test[i][cls], -0.5, 0.5))
- z = _zscore(proba[i])
-
- # ββ 2-day drawdown check ββββββββββββββββββββββββββββββββββββββββββββββ
- recent_rets.append(etf_ret)
- if len(recent_rets) > 2:
- recent_rets.pop(0)
- two_day = ((1 + recent_rets[0]) * (1 + recent_rets[-1]) - 1
- if len(recent_rets) >= 2 else 0.0)
-
- if two_day <= CASH_DRAWDOWN_TRIGGER:
- in_cash = True
- if in_cash and z >= CASH_EXIT_Z:
- in_cash = False
-
- # ββ Execute βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
- if in_cash:
- signal_etf = "CASH"
- realized_ret = daily_tbill
- else:
- signal_etf = etf_name
- realized_ret = etf_ret
-
- net_ret = realized_ret - fee
- strat_rets.append(net_ret)
- date_index.append(test_dates[i])
-
- if test_dates[i].date() < today:
- audit_trail.append({
- "Date": test_dates[i].strftime("%Y-%m-%d"),
- "Signal": signal_etf,
- "Net_Return": net_ret,
- "Z_Score": round(z, 2),
- })
-
- strat_rets = np.array(strat_rets, dtype=np.float64)
-
- # Next signal
- last_cls = min(int(preds[-1]), n_etfs - 1)
- last_z = _zscore(proba[-1])
- last_ret = float(np.clip(y_raw_test[-1][last_cls], -0.5, 0.5))
- prev_ret = float(np.clip(y_raw_test[-2][last_cls], -0.5, 0.5)) if len(y_raw_test) > 1 else 0.0
- last_2d = (1 + prev_ret) * (1 + last_ret) - 1
- next_cash = last_2d <= CASH_DRAWDOWN_TRIGGER and last_z < CASH_EXIT_Z
- next_signal = "CASH" if next_cash else target_etfs[last_cls].replace("_Ret", "")
-
- metrics = _compute_metrics(strat_rets, tbill_rate, date_index)
-
- return {
- **metrics,
- "strat_rets": strat_rets,
- "audit_trail": audit_trail,
- "next_signal": next_signal,
- "next_proba": proba[-1],
- }
-
-
-def _compute_metrics(strat_rets: np.ndarray, tbill_rate: float,
- date_index: list = None) -> dict:
- if len(strat_rets) == 0:
- return {}
-
- cum_returns = np.cumprod(1 + strat_rets)
- n = len(strat_rets)
- ann_return = float(cum_returns[-1] ** (252 / n) - 1)
-
- excess = strat_rets - tbill_rate / 252
- sharpe = float(np.mean(excess) / (np.std(strat_rets) + 1e-9) * np.sqrt(252))
-
- hit_ratio = float(np.mean(strat_rets[-15:] > 0))
-
- cum_max = np.maximum.accumulate(cum_returns)
- drawdown = (cum_returns - cum_max) / cum_max
- max_dd = float(np.min(drawdown))
-
- worst_idx = int(np.argmin(strat_rets))
- max_daily = float(strat_rets[worst_idx])
- worst_date = (date_index[worst_idx].strftime("%Y-%m-%d")
- if date_index and worst_idx < len(date_index) else "N/A")
-
- return {
- "cum_returns": cum_returns,
- "ann_return": ann_return,
- "sharpe": sharpe,
- "hit_ratio": hit_ratio,
- "max_dd": max_dd,
- "max_daily_dd": max_daily,
- "max_daily_date": worst_date,
- "cum_max": cum_max,
- }
-
-
-def compute_benchmark_metrics(returns: np.ndarray, tbill_rate: float) -> dict:
- return _compute_metrics(np.array(returns, dtype=np.float64), tbill_rate)
-
-
-def select_winner(results: dict) -> str:
- best_name, best_ret = None, -np.inf
- for name, res in results.items():
- if res is None:
- continue
- r = res.get("ann_return", -np.inf)
- if r > best_ret:
- best_ret, best_name = r, name
- return best_name
-
-
-def build_comparison_table(results: dict, winner_name: str) -> pd.DataFrame:
- rows = []
- for name, res in results.items():
- if res is None:
- rows.append({"Approach": name, "Ann. Return": "N/A",
- "Sharpe": "N/A", "Hit Ratio (15d)": "N/A",
- "Max Drawdown": "N/A", "Winner": ""})
+import hashlib
+import pickle
+from pathlib import Path
+from sklearn.preprocessing import RobustScaler
+from sklearn.utils.class_weight import compute_class_weight
+
+SEED = 42
+CACHE_DIR = Path("/tmp/p2_model_cache")
+CACHE_DIR.mkdir(exist_ok=True)
+np.random.seed(SEED)
+
+
+# ββ Cache helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def make_cache_key(last_date, start_yr, fee_bps, epochs, split, include_cash, lookback):
+ raw = f"{last_date}_{start_yr}_{fee_bps}_{epochs}_{split}_{include_cash}_{lookback}"
+ return hashlib.md5(raw.encode()).hexdigest()
+
+
+def save_cache(key, payload):
+ with open(CACHE_DIR / f"{key}.pkl", "wb") as f:
+ pickle.dump(payload, f)
+
+
+def load_cache(key):
+ path = CACHE_DIR / f"{key}.pkl"
+ if path.exists():
+ try:
+ with open(path, "rb") as f:
+ return pickle.load(f)
+ except Exception:
+ path.unlink(missing_ok=True)
+ return None
+
+
+# ββ Sequence builder ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def build_sequences(features, targets, lookback):
+ X, y = [], []
+ for i in range(lookback, len(features)):
+ X.append(features[i - lookback: i])
+ y.append(targets[i])
+ return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32)
+
+
+# ββ Train / val / test split ββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def train_val_test_split(X, y, train_pct=0.70, val_pct=0.15):
+ n = len(X)
+ t1 = int(n * train_pct)
+ t2 = int(n * (train_pct + val_pct))
+ return X[:t1], y[:t1], X[t1:t2], y[t1:t2], X[t2:], y[t2:]
+
+
+# ββ Feature scaling βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def scale_features(X_train, X_val, X_test):
+ n_feat = X_train.shape[2]
+ scaler = RobustScaler()
+ scaler.fit(X_train.reshape(-1, n_feat))
+ def _t(X):
+ s = X.shape
+ return scaler.transform(X.reshape(-1, n_feat)).reshape(s)
+ return _t(X_train), _t(X_val), _t(X_test), scaler
+
+
+# ββ Label builder (no CASH class β CASH is a risk overlay) βββββββββββββββββββ
+
+def returns_to_labels(y_raw):
+ """Simple argmax β model always predicts one of the ETFs."""
+ return np.argmax(y_raw, axis=1).astype(np.int32)
+
+
+# ββ Class weights βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def compute_class_weights(y_labels, n_classes):
+ present = np.unique(y_labels)
+ try:
+ weights = compute_class_weight("balanced", classes=present, y=y_labels)
+ weight_dict = {int(c): float(w) for c, w in zip(present, weights)}
+ except Exception:
+ weight_dict = {}
+ for c in range(n_classes):
+ if c not in weight_dict:
+ weight_dict[c] = 1.0
+ return weight_dict
+
+
+# ββ Callbacks βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def get_callbacks(patience_es=15, patience_lr=8, min_lr=1e-6):
+ from tensorflow import keras
+ return [
+ keras.callbacks.EarlyStopping(
+ monitor="val_loss", patience=patience_es,
+ restore_best_weights=True, verbose=0,
+ ),
+ keras.callbacks.ReduceLROnPlateau(
+ monitor="val_loss", factor=0.5,
+ patience=patience_lr, min_lr=min_lr, verbose=0,
+ ),
+ ]
+
+
+# ββ Output head βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+def classification_head(x, n_classes, dropout=0.3):
+ from tensorflow import keras
+ x = keras.layers.Dense(32, activation="relu")(x)
+ x = keras.layers.Dropout(dropout)(x)
+ x = keras.layers.Dense(n_classes, activation="softmax")(x)
+ return x
+
+
+# ββ Auto lookback selection (Approach 1 proxy, fast) βββββββββββββββββββββββββ
+
+def find_best_lookback(X_raw, y_raw, train_pct, val_pct, n_classes,
+ candidates=None):
+ from tensorflow import keras
+
+ if candidates is None:
+ candidates = [30, 45, 60]
+
+ best_lb, best_loss = candidates[0], np.inf
+
+ for lb in candidates:
+ try:
+ X_seq, y_seq = build_sequences(X_raw, y_raw, lb)
+ y_lab = returns_to_labels(y_seq)
+ X_tr, y_tr, X_v, y_v, _, _ = train_val_test_split(X_seq, y_lab, train_pct, val_pct)
+ X_tr_s, X_v_s, _, _ = scale_features(X_tr, X_v, X_v)
+ cw = compute_class_weights(y_tr, n_classes)
+
+ inp = keras.Input(shape=X_tr_s.shape[1:])
+ x = keras.layers.Conv1D(16, min(3, lb), padding="causal", activation="relu")(inp)
+ x = keras.layers.GlobalAveragePooling1D()(x)
+ out = keras.layers.Dense(n_classes, activation="softmax")(x)
+ m = keras.Model(inp, out)
+ m.compile(optimizer="adam", loss="sparse_categorical_crossentropy")
+
+ hist = m.fit(
+ X_tr_s, y_tr,
+ validation_data=(X_v_s, y_v),
+ epochs=15, batch_size=64, class_weight=cw,
+ callbacks=[keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True)],
+ verbose=0,
+ )
+ val_loss = min(hist.history.get("val_loss", [np.inf]))
+ if val_loss < best_loss:
+ best_loss, best_lb = val_loss, lb
+ del m
+ except Exception:
continue
- rows.append({
- "Approach": name,
- "Ann. Return": f"{res['ann_return']*100:.2f}%",
- "Sharpe": f"{res['sharpe']:.2f}",
- "Hit Ratio (15d)": f"{res['hit_ratio']*100:.0f}%",
- "Max Drawdown": f"{res['max_dd']*100:.2f}%",
- "Winner": "β WINNER" if name == winner_name else "",
- })
- return pd.DataFrame(rows)
+
+ return best_lb
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py
index 5f062a870f25bb5e93ede2fbc2d52de6aff150c0..af700cba7a4ef3f6441a9647158f9d7115df6af0 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/ui/components.py
@@ -76,6 +76,25 @@ def show_all_signals_panel(all_signals: dict, target_etfs: list,
""", unsafe_allow_html=True)
+def _build_etf_badges(sorted_pairs: list, best_name: str, color: str) -> str:
+ """Build compact ETF probability badges as HTML string."""
+ badges = []
+ for name, score in sorted_pairs:
+ is_best = name == best_name
+ bg = "#e8fdf7" if is_best else "#f8f8f8"
+ border = color if is_best else "#ddd"
+ txt_col = color if is_best else "#555"
+ weight = "700" if is_best else "400"
+ star = "β
" if is_best else ""
+ badges.append(
+ f'
'
+ f'{star}{name} {score:.3f}'
+ )
+ return "".join(badges)
+
+
# ββ Signal conviction panel βββββββββββββββββββββββββββββββββββββββββββββββββββ
def show_conviction_panel(conviction: dict):
@@ -121,14 +140,7 @@ def show_conviction_panel(conviction: dict):
MODEL PROBABILITY BY ETF
- {"".join([
- f''
- f'{"β
" if n == best_name else ""}{n} {s:.3f}'
- for n, s in sorted_pairs
- ])}
+ {_build_etf_badges(sorted_pairs, best_name, color)}
""", unsafe_allow_html=True)
diff --git a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
index 36fe25e40e923799b2a57e008d8aba1d83178e3d..17780796b4e1482fc78b2965712426d2e53ad3fc 100644
--- a/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
+++ b/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/hf_space/models/base.py
@@ -122,7 +122,7 @@ def classification_head(x, n_classes, dropout=0.3):
# ββ Auto lookback selection (Approach 1 proxy, fast) βββββββββββββββββββββββββ
def find_best_lookback(X_raw, y_raw, train_pct, val_pct, n_classes,
- candidates=None):
+ include_cash=False, candidates=None):
from tensorflow import keras
if candidates is None:
diff --git a/hf_space/hf_space/hf_space/hf_space/models/base.py b/hf_space/hf_space/hf_space/hf_space/models/base.py
index 17780796b4e1482fc78b2965712426d2e53ad3fc..4fe679ed1a7ee751931f365fcacae60b546c593a 100644
--- a/hf_space/hf_space/hf_space/hf_space/models/base.py
+++ b/hf_space/hf_space/hf_space/hf_space/models/base.py
@@ -20,7 +20,7 @@ np.random.seed(SEED)
# ββ Cache helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def make_cache_key(last_date, start_yr, fee_bps, epochs, split, include_cash, lookback):
- raw = f"{last_date}_{start_yr}_{fee_bps}_{epochs}_{split}_{include_cash}_{lookback}"
+ raw = f"v2_{last_date}_{start_yr}_{fee_bps}_{epochs}_{split}_{include_cash}_{lookback}"
return hashlib.md5(raw.encode()).hexdigest()
diff --git a/hf_space/hf_space/hf_space/models/base.py b/hf_space/hf_space/hf_space/models/base.py
index 4fe679ed1a7ee751931f365fcacae60b546c593a..b178b8c7a5cc95607c2545447da03ceb360a6d3b 100644
--- a/hf_space/hf_space/hf_space/models/base.py
+++ b/hf_space/hf_space/hf_space/models/base.py
@@ -14,6 +14,23 @@ from sklearn.utils.class_weight import compute_class_weight
SEED = 42
CACHE_DIR = Path("/tmp/p2_model_cache")
CACHE_DIR.mkdir(exist_ok=True)
+
+# Clear any v1 cache files (missing max_daily_date field)
+for _f in CACHE_DIR.glob("*.pkl"):
+ try:
+ import pickle as _pkl
+ with open(_f, "rb") as _fh:
+ _d = _pkl.load(_fh)
+ # If any result dict lacks max_daily_date, bust the whole cache
+ if isinstance(_d, dict) and "results" in _d:
+ _needs_bust = any(
+ isinstance(r, dict) and "max_daily_date" not in r
+ for r in _d["results"].values() if r is not None
+ )
+ if _needs_bust:
+ _f.unlink(missing_ok=True)
+ except Exception:
+ _f.unlink(missing_ok=True)
np.random.seed(SEED)
diff --git a/hf_space/hf_space/ui/components.py b/hf_space/hf_space/ui/components.py
index af700cba7a4ef3f6441a9647158f9d7115df6af0..c5dd92131d74943e3b1daaff0d2ada49f634fed2 100644
--- a/hf_space/hf_space/ui/components.py
+++ b/hf_space/hf_space/ui/components.py
@@ -173,10 +173,10 @@ def show_metrics_row(result: dict, tbill_rate: float, spy_ann_return: float = No
c4.metric("π Max Drawdown", f"{result['max_dd']*100:.2f}%",
delta="Peak to Trough")
- # Max daily DD with date
+ # Max daily DD with date (only show date if available)
worst_date = result.get("max_daily_date", "N/A")
- c5.metric("β οΈ Max Daily DD", f"{result['max_daily_dd']*100:.2f}%",
- delta=f"on {worst_date}")
+ dd_delta = f"on {worst_date}" if worst_date != "N/A" else "Worst Single Day"
+ c5.metric("β οΈ Max Daily DD", f"{result['max_daily_dd']*100:.2f}%", delta=dd_delta)
# ββ Comparison table ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ