r""" regime_classifier_v1.py ======================== Macro-state regime classifier for EURUSD and USDJPY. Classifies each day based on yield spread DISLOCATION (level + velocity), not on whether signals have fired. This means the tier reflects the MACRO ENVIRONMENT, not the strategy's reaction to it. Two consequences: 1. The tier explains WHY signals are firing (or not). 2. Dislocated macro that hasn't yet produced signals still classifies high — a useful early indicator. EURUSD: 5 tiers, 14-day spread velocity window - High Dislocation (top 10%) - Elevated (60-90th pct) - Normal (30-60th pct) - Subdued (10-30th pct) - Compressed (bottom 10%) USDJPY: 3 tiers, 60-day spread velocity window - Active (top 25%) - Normal (25-75th pct) - Subdued (bottom 25%) INPUTS per pair: - Spread LEVEL z-score (vs full history) - Spread VELOCITY z-score (N-day rate of change, vs full history) Composite = |velocity_z| + 0.5 * |level_z| Higher composite = more macro dislocation OUTPUTS: data/research/regime_thresholds.json -- cutoff numbers for monitor/dashboard data/research/regime_history_eurusd.md -- historical tier report data/research/regime_history_usdjpy.md -- same for UJ Console: current tier + sanity-check on known macro periods Run from project root: cd C:\\Users\\paul_\\OneDrive\\fx_macro_intraday python src\\research\\regime_classifier_v1.py """ from __future__ import annotations import json from pathlib import Path from datetime import datetime import numpy as np import pandas as pd BASE = Path(__file__).resolve().parents[2] RATES_DIR = BASE / "data" / "raw" / "rates" OUT_DIR = BASE / "data" / "research" OUT_DIR.mkdir(parents=True, exist_ok=True) # Configuration EU_VELOCITY_DAYS = 14 UJ_VELOCITY_DAYS = 60 LEVEL_WEIGHT = 0.5 EU_PERCENTILES = [10, 30, 60, 90] UJ_PERCENTILES = [25, 75] EU_TIER_NAMES = ["Compressed", "Subdued", "Normal", "Elevated", "High Dislocation"] UJ_TIER_NAMES = ["Subdued", "Normal", "Active"] # Activity tier configuration (signal density) EU_ACTIVITY_WINDOW = 14 # days UJ_ACTIVITY_WINDOW = 60 # days EU_ACTIVITY_PERCENTILES = [10, 30, 60, 90] UJ_ACTIVITY_PERCENTILES = [25, 75] EU_ACTIVITY_NAMES = ["Quiet", "Light", "Steady", "Active", "Peak"] UJ_ACTIVITY_NAMES = ["Quiet", "Steady", "Active"] TRADES_DIR_EU = BASE / "data" / "processed" / "trades" / "trades_real_costs.csv" TRADES_DIR_UJ = BASE / "data" / "processed" / "usdjpy_trades_real_costs.csv" SANITY_CHECKS_EU = [ ("2008-09-15", "2008-12-31", "Lehman / GFC peak"), ("2015-01-01", "2015-06-30", "ECB QE launch"), ("2017-06-01", "2017-12-31", "Calm 2017"), ("2020-03-01", "2020-06-30", "COVID shock"), ("2022-01-01", "2022-12-31", "Fed hiking cycle"), ("2024-01-01", "2024-12-31", "Recent regime 2024"), ("2025-01-01", "2025-12-31", "Recent regime 2025"), ("2026-01-01", "2026-05-13", "Current YTD"), ] SANITY_CHECKS_UJ = [ ("2008-09-15", "2008-12-31", "Lehman / GFC peak"), ("2013-04-01", "2013-12-31", "BoJ Abenomics QE"), ("2022-09-22", "2022-12-31", "JPY intervention 2022"), ("2024-01-01", "2024-12-31", "Recent regime 2024"), ("2026-01-01", "2026-05-13", "Current YTD"), ] def load_rate(fname, col): p = RATES_DIR / fname if not p.exists(): raise FileNotFoundError(f"Rate CSV not found: {p}") df = pd.read_csv(p) df.columns = [c.lower() for c in df.columns] df["date"] = pd.to_datetime(df["date"]) df[col] = pd.to_numeric(df[col], errors="coerce") return df[["date", col]].dropna().sort_values("date").reset_index(drop=True) def build_spread_series(a_csv, a_col, b_csv, b_col): a = load_rate(a_csv, a_col) b = load_rate(b_csv, b_col) dr = pd.DataFrame({"date": pd.date_range( min(a["date"].min(), b["date"].min()), max(a["date"].max(), b["date"].max()), freq="D")}) a = pd.merge(dr, a, on="date", how="left").ffill() b = pd.merge(dr, b, on="date", how="left").ffill() df = pd.merge(a, b, on="date", how="inner").dropna() df["spread"] = df[a_col] - df[b_col] return df[["date", "spread"]].reset_index(drop=True) def compute_composite_score(spread_df, velocity_days): df = spread_df.copy() df["velocity"] = df["spread"] - df["spread"].shift(velocity_days) warmup = velocity_days + 30 lm = df["spread"].iloc[warmup:].mean() ls = df["spread"].iloc[warmup:].std() vm = df["velocity"].iloc[warmup:].mean() vs = df["velocity"].iloc[warmup:].std() df["level_z"] = (df["spread"] - lm) / ls df["velocity_z"] = (df["velocity"] - vm) / vs df["composite"] = df["velocity_z"].abs() + LEVEL_WEIGHT * df["level_z"].abs() return df def calibrate_cutoffs(score_series, percentile_bounds): s = score_series.dropna().values if len(s) < 200: raise RuntimeError("Not enough data to calibrate cutoffs") return [float(np.percentile(s, p)) for p in percentile_bounds] def assign_tier(score, cutoffs, tier_names): if score is None or (isinstance(score, float) and np.isnan(score)): return "Unknown" for i, c in enumerate(cutoffs): if score < c: return tier_names[i] return tier_names[-1] def calibrate_activity_cutoffs(trades_csv, window_days, percentiles): """Calibrate signal-density tier cutoffs from full historical trade table. Computes a rolling N-day signal count across the full validated trade history, then returns the percentile cutoffs that define tier boundaries. Returns (cutoffs, n_trades, date_range_str) tuple, or (None, 0, "") on failure. """ if not trades_csv.exists(): return None, 0, "" df = pd.read_csv(trades_csv) if "entry_time" not in df.columns or df.empty: return None, 0, "" df["entry_time"] = pd.to_datetime(df["entry_time"]) df = df.sort_values("entry_time").reset_index(drop=True) start = df["entry_time"].min().normalize() end = df["entry_time"].max().normalize() all_days = pd.date_range(start, end, freq="D") entry_dates = df["entry_time"].dt.normalize() counts = np.zeros(len(all_days), dtype=int) for i, d in enumerate(all_days): win_start = d - pd.Timedelta(days=window_days) counts[i] = int(((entry_dates > win_start) & (entry_dates <= d)).sum()) # Skip warmup (first window_days of history) counts = counts[window_days:] if len(counts) < 100: return None, 0, "" cutoffs = [float(np.percentile(counts, p)) for p in percentiles] return cutoffs, len(df), f"{start.date()} -> {end.date()}" def build_history_report(df, cutoffs, tier_names, pair_label, velocity_days, sanity_checks): df = df.copy() df["tier"] = df["composite"].apply(lambda s: assign_tier(s, cutoffs, tier_names)) df["year"] = df["date"].dt.year valid = df.dropna(subset=["composite"]) pivot = valid.groupby(["year", "tier"]).size().unstack(fill_value=0) pivot = pivot.reindex(columns=tier_names, fill_value=0) pivot_pct = pivot.div(pivot.sum(axis=1), axis=0) * 100 lines = [ f"# {pair_label} Macro Regime History Report", "", f"**Calibrated:** {datetime.now().isoformat(timespec='seconds')}", f"**Velocity lookback:** {velocity_days} days", f"**Score formula:** |velocity_z| + {LEVEL_WEIGHT} * |level_z|", "", "## Tier cutoffs (composite score)", "", ] for tier_name, lo, hi in zip(tier_names, [None] + cutoffs, cutoffs + [None]): if lo is None: lines.append(f"- **{tier_name}**: score < {hi:.3f}") elif hi is None: lines.append(f"- **{tier_name}**: score >= {lo:.3f}") else: lines.append(f"- **{tier_name}**: {lo:.3f} <= score < {hi:.3f}") lines.append("") lines.append(f"## Coverage over {len(valid)} valid days") lines.append("") coverage = (valid["tier"].value_counts(normalize=True) * 100).reindex(tier_names, fill_value=0) for tier_name, pct in coverage.items(): lines.append(f"- {tier_name}: {pct:.1f}% of days") lines.append("") lines.append("## Per-year tier dominance (% of days)") lines.append("") lines.append("```") lines.append(pivot_pct.round(1).to_string()) lines.append("```") lines.append("") lines.append("## Sanity-check periods") lines.append("") for start, end, label in sanity_checks: mask = (df["date"] >= start) & (df["date"] <= end) period = df[mask].dropna(subset=["composite"]) if period.empty: lines.append(f"- **{label}** ({start} -> {end}): no data") continue modal_tier = period["tier"].mode().iloc[0] avg_score = period["composite"].mean() tier_dist = (period["tier"].value_counts(normalize=True) * 100) dist_str = ", ".join(f"{t} {p:.0f}%" for t, p in tier_dist.items() if p >= 5) lines.append(f"- **{label}** ({start} -> {end}): dominant **{modal_tier}** " f"(avg score {avg_score:.2f}) -- {dist_str}") lines.append("") lines.append(f"## Continuous tier runs >= 60 days (highest/lowest tiers only)") lines.append("") df_sorted = valid.sort_values("date").reset_index(drop=True) df_sorted["tier_change"] = (df_sorted["tier"] != df_sorted["tier"].shift()).cumsum() runs = (df_sorted.groupby(["tier_change", "tier"]) .agg(start=("date", "min"), end=("date", "max"), n_days=("date", "count")) .reset_index()) long_runs = runs[runs["n_days"] >= 60].sort_values("start") for highlight in [tier_names[-1], tier_names[0]]: rows = long_runs[long_runs["tier"] == highlight] if not rows.empty: lines.append(f"### {highlight} runs") for _, r in rows.iterrows(): lines.append(f"- {r['start'].date()} -> {r['end'].date()} ({r['n_days']} days)") lines.append("") return "\n".join(lines) def main(): print("=" * 70) print(" MACRO REGIME CLASSIFIER v1 -- Calibration") print("=" * 70) print(f" Run time: {datetime.now()}") print() # EURUSD print("EURUSD -- Loading US-DE 2Y spread...") eu_spread = build_spread_series("us2y.csv", "us2y", "de2y.csv", "de2y") print(f" Spread series: {len(eu_spread):,} days " f"({eu_spread['date'].min().date()} -> {eu_spread['date'].max().date()})") eu_df = compute_composite_score(eu_spread, EU_VELOCITY_DAYS) print(f" Valid composite days: {len(eu_df.dropna(subset=['composite'])):,}") eu_cutoffs = calibrate_cutoffs(eu_df["composite"], EU_PERCENTILES) print(f" Cutoffs at percentiles {EU_PERCENTILES}: " f"{[round(c, 3) for c in eu_cutoffs]}") eu_report = build_history_report( eu_df, eu_cutoffs, EU_TIER_NAMES, "EURUSD", EU_VELOCITY_DAYS, SANITY_CHECKS_EU) eu_path = OUT_DIR / "regime_history_eurusd.md" eu_path.write_text(eu_report, encoding="utf-8") print(f" Report saved: {eu_path}") # USDJPY print() print("USDJPY -- Loading US-JP 2Y spread...") uj_spread = build_spread_series("us2y.csv", "us2y", "jp2y.csv", "jp2y") print(f" Spread series: {len(uj_spread):,} days " f"({uj_spread['date'].min().date()} -> {uj_spread['date'].max().date()})") uj_df = compute_composite_score(uj_spread, UJ_VELOCITY_DAYS) print(f" Valid composite days: {len(uj_df.dropna(subset=['composite'])):,}") uj_cutoffs = calibrate_cutoffs(uj_df["composite"], UJ_PERCENTILES) print(f" Cutoffs at percentiles {UJ_PERCENTILES}: " f"{[round(c, 3) for c in uj_cutoffs]}") uj_report = build_history_report( uj_df, uj_cutoffs, UJ_TIER_NAMES, "USDJPY", UJ_VELOCITY_DAYS, SANITY_CHECKS_UJ) uj_path = OUT_DIR / "regime_history_usdjpy.md" uj_path.write_text(uj_report, encoding="utf-8") print(f" Report saved: {uj_path}") # ── Activity (signal density) cutoffs ── print() print("Calibrating ACTIVITY cutoffs from validated trade history...") eu_act_cutoffs, eu_act_n, eu_act_range = calibrate_activity_cutoffs( TRADES_DIR_EU, EU_ACTIVITY_WINDOW, EU_ACTIVITY_PERCENTILES) if eu_act_cutoffs: print(f" EURUSD ({EU_ACTIVITY_WINDOW}d window, {eu_act_n} trades, {eu_act_range}):") print(f" Cutoffs at percentiles {EU_ACTIVITY_PERCENTILES}: " f"{[round(c, 1) for c in eu_act_cutoffs]}") else: print(f" EURUSD: could not calibrate (trade CSV missing or too small)") uj_act_cutoffs, uj_act_n, uj_act_range = calibrate_activity_cutoffs( TRADES_DIR_UJ, UJ_ACTIVITY_WINDOW, UJ_ACTIVITY_PERCENTILES) if uj_act_cutoffs: print(f" USDJPY ({UJ_ACTIVITY_WINDOW}d window, {uj_act_n} trades, {uj_act_range}):") print(f" Cutoffs at percentiles {UJ_ACTIVITY_PERCENTILES}: " f"{[round(c, 1) for c in uj_act_cutoffs]}") else: print(f" USDJPY: could not calibrate (trade CSV missing or too small)") # Save thresholds thresholds = { "calibrated_at": datetime.now().isoformat(), "score_formula": f"|velocity_z| + {LEVEL_WEIGHT} * |level_z|", "eurusd": { "velocity_days": EU_VELOCITY_DAYS, "tier_names": EU_TIER_NAMES, "cutoffs": eu_cutoffs, "percentile_bounds": EU_PERCENTILES, "spread": "US 2Y - DE 2Y", "activity_window_days": EU_ACTIVITY_WINDOW, "activity_tier_names": EU_ACTIVITY_NAMES, "activity_cutoffs": eu_act_cutoffs, "activity_percentiles": EU_ACTIVITY_PERCENTILES, }, "usdjpy": { "velocity_days": UJ_VELOCITY_DAYS, "tier_names": UJ_TIER_NAMES, "cutoffs": uj_cutoffs, "percentile_bounds": UJ_PERCENTILES, "spread": "US 2Y - JP 2Y", "activity_window_days": UJ_ACTIVITY_WINDOW, "activity_tier_names": UJ_ACTIVITY_NAMES, "activity_cutoffs": uj_act_cutoffs, "activity_percentiles": UJ_ACTIVITY_PERCENTILES, }, } t_path = OUT_DIR / "regime_thresholds.json" t_path.write_text(json.dumps(thresholds, indent=2), encoding="utf-8") print() print(f"Thresholds JSON saved: {t_path}") # Current tier print() print("=" * 70) print(" CURRENT TIER (latest rate data)") print("=" * 70) eu_latest = eu_df.dropna(subset=["composite"]).iloc[-1] eu_now = assign_tier(eu_latest["composite"], eu_cutoffs, EU_TIER_NAMES) print(f"\n EURUSD: {eu_now}") print(f" Date: {eu_latest['date'].date()}") print(f" Spread: {eu_latest['spread']:.4f} (level_z={eu_latest['level_z']:+.2f})") print(f" Velocity: {eu_latest['velocity']:+.4f} over {EU_VELOCITY_DAYS}d " f"(velocity_z={eu_latest['velocity_z']:+.2f})") print(f" Composite: {eu_latest['composite']:.3f}") uj_latest = uj_df.dropna(subset=["composite"]).iloc[-1] uj_now = assign_tier(uj_latest["composite"], uj_cutoffs, UJ_TIER_NAMES) print(f"\n USDJPY: {uj_now}") print(f" Date: {uj_latest['date'].date()}") print(f" Spread: {uj_latest['spread']:.4f} (level_z={uj_latest['level_z']:+.2f})") print(f" Velocity: {uj_latest['velocity']:+.4f} over {UJ_VELOCITY_DAYS}d " f"(velocity_z={uj_latest['velocity_z']:+.2f})") print(f" Composite: {uj_latest['composite']:.3f}") # Sanity check print() print("=" * 70) print(" SANITY CHECK") print("=" * 70) print("\nEURUSD:") for start, end, label in SANITY_CHECKS_EU: mask = (eu_df["date"] >= start) & (eu_df["date"] <= end) p = eu_df[mask].dropna(subset=["composite"]) if p.empty: print(f" {label:<45} no data") continue p = p.copy() p["tier"] = p["composite"].apply(lambda s: assign_tier(s, eu_cutoffs, EU_TIER_NAMES)) print(f" {label:<45} {p['tier'].mode().iloc[0]:<20} avg score {p['composite'].mean():.2f}") print("\nUSDJPY:") for start, end, label in SANITY_CHECKS_UJ: mask = (uj_df["date"] >= start) & (uj_df["date"] <= end) p = uj_df[mask].dropna(subset=["composite"]) if p.empty: print(f" {label:<45} no data") continue p = p.copy() p["tier"] = p["composite"].apply(lambda s: assign_tier(s, uj_cutoffs, UJ_TIER_NAMES)) print(f" {label:<45} {p['tier'].mode().iloc[0]:<20} avg score {p['composite'].mean():.2f}") # Coverage print() print("=" * 70) print(" COVERAGE -- % of valid days in each tier") print("=" * 70) eu_v = eu_df.dropna(subset=["composite"]).copy() eu_v["tier"] = eu_v["composite"].apply(lambda s: assign_tier(s, eu_cutoffs, EU_TIER_NAMES)) print("\n EURUSD:") for t in EU_TIER_NAMES: pct = (eu_v["tier"] == t).mean() * 100 print(f" {t:<20} {pct:>5.1f}%") uj_v = uj_df.dropna(subset=["composite"]).copy() uj_v["tier"] = uj_v["composite"].apply(lambda s: assign_tier(s, uj_cutoffs, UJ_TIER_NAMES)) print("\n USDJPY:") for t in UJ_TIER_NAMES: pct = (uj_v["tier"] == t).mean() * 100 print(f" {t:<20} {pct:>5.1f}%") print() print("Done.") if __name__ == "__main__": main()