r""" zscore_recent_check.py (v2 — exact beta model) ================================================ Replicates the EXACT live monitor signal computation for both pairs and shows when the z-score would have breached threshold in the last 8 weeks. EURUSD: Rolling-beta lag-gap z-score (120-day OLS, 20-day EWM smooth, 60-bar z-window) USDJPY: US-JP 2Y spread level z-score (30-day window) Run on home machine: cd C:/Users/paul_/OneDrive/fx_macro_intraday python src\research\zscore_recent_check.py """ import warnings, sys from pathlib import Path import numpy as np import pandas as pd warnings.filterwarnings('ignore') BASE = Path(__file__).resolve().parents[2] RATES_DIR = BASE / 'data' / 'raw' / 'rates' PRICE_DIR = BASE / 'data' / 'raw' / 'prices' # ── Parameters (exact match to live monitor) ────────────────────────────────── EU_THRESHOLD = 2.75 UJ_THRESHOLD = 2.00 BETA_WINDOW = 120 SMOOTH_SPAN = 20 EU_Z_WINDOW = 60 UJ_Z_WINDOW = 30 # ── Helpers ─────────────────────────────────────────────────────────────────── def load_rate(fname, col): p = RATES_DIR / fname if not p.exists(): print(f" ⚠️ Missing: {p}") return None df = pd.read_csv(p) df.columns = [c.lower() for c in df.columns] df['date'] = pd.to_datetime(df.iloc[:,0], errors='coerce') df[col] = pd.to_numeric(df.iloc[:,1], errors='coerce') return df[['date', col]].dropna().sort_values('date') def load_price_1h(symbol): """ Load 1H price data. Joins historical + recent files for full coverage. """ def _read(p): df = pd.read_csv(p) df.columns = [c.lower() for c in df.columns] for c in ['datetime','date','time','timestamp']: if c in df.columns: df['datetime'] = pd.to_datetime(df[c], errors='coerce') break if 'datetime' not in df.columns: df['datetime'] = pd.to_datetime(df.iloc[:,0], errors='coerce') df['close'] = pd.to_numeric(df['close'], errors='coerce') return df.dropna(subset=['datetime','close'])[['datetime','close']].sort_values('datetime') hist_path = PRICE_DIR / symbol / f"{symbol}_1H_2003_2026.csv" recent_path = PRICE_DIR / symbol / f"{symbol}_1H_recent.csv" frames = [] for p in [hist_path, recent_path]: if p.exists(): try: df = _read(p) frames.append(df) print(f" Loaded: {p.name} ({len(df):,} bars | " f"{df['datetime'].iloc[0].date()} to {df['datetime'].iloc[-1].date()})") except Exception as e: print(f" Could not read {p.name}: {e}") if not frames: print(f" No 1H price files found for {symbol}") return None combined = (pd.concat(frames) .drop_duplicates('datetime') .sort_values('datetime') .reset_index(drop=True)) print(f" Combined: {len(combined):,} bars | " f"{combined['datetime'].iloc[0].date()} to {combined['datetime'].iloc[-1].date()}") return combined print("=" * 70) print(" RECENT Z-SCORE CHECK v2 — Exact Beta Model") print("=" * 70) # ══════════════════════════════════════════════════════════════════════════════ # EURUSD — Full Beta Model # ══════════════════════════════════════════════════════════════════════════════ print(f"\n{'─'*70}") print(" EURUSD — Rolling Beta Lag-Gap Z-Score (exact live model)") print(f" Beta window: {BETA_WINDOW}d | EWM smooth: {SMOOTH_SPAN} | Z-window: {EU_Z_WINDOW}h | Threshold: ±{EU_THRESHOLD}") print(f"{'─'*70}") try: import statsmodels.api as sm prices_1h = load_price_1h('EURUSD') if prices_1h is None: raise ValueError("No EURUSD price data") us2y = load_rate('us2y.csv', 'us2y') us10y = load_rate('us10y.csv', 'us10y') de2y = load_rate('de2y.csv', 'de2y') de10y = load_rate('de10y.csv', 'de10y') if any(x is None for x in [us2y, us10y, de2y, de10y]): raise ValueError("Missing rate files") prices_1h = prices_1h.copy() prices_1h['date'] = prices_1h['datetime'].dt.normalize() daily_px = (prices_1h.groupby('date', as_index=False) .agg(close=('close','last')) .sort_values('date').reset_index(drop=True)) daily_px['eurusd_return_1d'] = daily_px['close'].pct_change() daily_px['date'] = pd.to_datetime(daily_px['date']) date_range = pd.DataFrame({'date': pd.date_range( daily_px['date'].min(), daily_px['date'].max(), freq='D')}) def ffill(rate_df, col): left = date_range.copy().astype({'date': 'datetime64[us]'}) right = rate_df.copy().astype({'date': 'datetime64[us]'}) return pd.merge_asof(left, right, on='date', direction='backward') spreads = (ffill(us2y,'us2y') .merge(ffill(us10y,'us10y'), on='date') .merge(ffill(de2y,'de2y'), on='date') .merge(ffill(de10y,'de10y'), on='date')) spreads['spread_2y'] = spreads['us2y'] - spreads['de2y'] spreads['spread_10y'] = spreads['us10y'] - spreads['de10y'] spreads['spread_2y_change_1d'] = spreads['spread_2y'].diff(1) spreads['spread_10y_change_1d'] = spreads['spread_10y'].diff(1) spreads = spreads[['date','spread_2y_change_1d','spread_10y_change_1d']].dropna() model_df = daily_px.merge(spreads, on='date', how='left').dropna( subset=['eurusd_return_1d','spread_2y_change_1d']) print(f"\n Running OLS on {len(model_df)} daily bars...") n = len(model_df) b2y_raw = np.full(n, np.nan) b10y_raw= np.full(n, np.nan) for i in range(BETA_WINDOW, n): sample = model_df.iloc[i-BETA_WINDOW:i].copy() sample = sample[(sample['spread_2y_change_1d'].abs()>0) | (sample['spread_10y_change_1d'].abs()>0)] if len(sample) < 30: continue X = sm.add_constant(sample[['spread_2y_change_1d','spread_10y_change_1d']]) try: res = sm.OLS(sample['eurusd_return_1d'], X).fit() b2y_raw[i] = res.params.get('spread_2y_change_1d', np.nan) b10y_raw[i] = res.params.get('spread_10y_change_1d', np.nan) except Exception: pass model_df['beta_2y'] = pd.Series(b2y_raw).ewm(span=SMOOTH_SPAN).mean().shift(1).clip(-0.10, 0.10).values model_df['beta_10y'] = pd.Series(b10y_raw).ewm(span=SMOOTH_SPAN).mean().shift(1).clip(-0.08, 0.08).values model_df['predicted_return_1d'] = ( model_df['beta_2y'] * model_df['spread_2y_change_1d'] + model_df['beta_10y'] * model_df['spread_10y_change_1d']) model_df = model_df[['date','predicted_return_1d']] df = prices_1h.merge(model_df, on='date', how='left') df['predicted_return_1d'] = df['predicted_return_1d'].ffill() df['eurusd_return_24h'] = df['close'].pct_change(24) df['lag_gap_24h'] = df['predicted_return_1d'] - df['eurusd_return_24h'] mean = df['lag_gap_24h'].rolling(EU_Z_WINDOW, min_periods=EU_Z_WINDOW).mean() std = df['lag_gap_24h'].rolling(EU_Z_WINDOW, min_periods=EU_Z_WINDOW).std() df['zscore'] = (df['lag_gap_24h'] - mean) / std df = df.dropna(subset=['zscore']) # Filter to last 8 weeks cutoff = pd.Timestamp.now() - pd.Timedelta(weeks=8) recent = df[df['datetime'] >= cutoff].copy() # In-session filter: EURUSD 06:00-15:00 UTC recent['hour_utc'] = recent['datetime'].dt.hour in_session = recent[(recent['hour_utc'] >= 6) & (recent['hour_utc'] <= 15)] print(f"\n Showing: in-session hours only (06:00-15:00 UTC)") print(f" All threshold breaches shown + daily in-session peak") # Show every in-session breach + daily in-session peak in_session = in_session.copy() in_session['date'] = in_session['datetime'].dt.date # Daily in-session peak daily_peak = (in_session.groupby('date') .apply(lambda x: x.loc[x['zscore'].abs().idxmax()]) [['datetime','zscore']] .reset_index(drop=True)) print(f"\n {'Date':>12} {'Time (UTC)':>10} {'Z-Score':>8} {'Status':>25}") print(f" {'─'*60}") signals_eu = 0 for _, row in daily_peak.iterrows(): z = row['zscore'] dt = row['datetime'] if abs(z) >= EU_THRESHOLD: status = f"🚨 SIGNAL ({'LONG' if z>0 else 'SHORT'})" signals_eu += 1 elif abs(z) >= 2.0: status = f"⚠️ Approaching ({abs(z):.2f})" else: status = "No signal" marker = " ◄ BREACH" if abs(z) >= EU_THRESHOLD else "" print(f" {str(dt.date()):>12} {str(dt.time())[:5]:>10} {z:>+8.3f} {status}{marker}") print(f"\n Summary (in-session hours only, 06:00-15:00 UTC):") print(f" Z-score range: {in_session['zscore'].min():.3f} to {in_session['zscore'].max():.3f}") print(f" Max |z| in session: {in_session['zscore'].abs().max():.3f}") print(f" In-session threshold breaches (±{EU_THRESHOLD}): {signals_eu}") if signals_eu == 0: dist = EU_THRESHOLD - in_session['zscore'].abs().max() print(f" Closest in-session approach: {in_session['zscore'].abs().max():.3f} ({dist:.3f} from threshold)") except Exception as e: print(f" ❌ EURUSD beta model failed: {e}") import traceback; traceback.print_exc() # ══════════════════════════════════════════════════════════════════════════════ # USDJPY — Spread Level Z-Score (exact, same as live) # ══════════════════════════════════════════════════════════════════════════════ print(f"\n{'─'*70}") print(" USDJPY — Spread Level Z-Score (exact live model)") print(f" Z-window: {UJ_Z_WINDOW}d | Threshold: ±{UJ_THRESHOLD}") print(f"{'─'*70}") try: us2y = load_rate('us2y.csv', 'us2y') jp2y = load_rate('jp2y.csv', 'jp2y') if us2y is None or jp2y is None: raise ValueError("Missing rate files") uj = pd.merge(us2y, jp2y, on='date', how='outer').sort_values('date') uj = uj.set_index('date') uj = uj.reindex(pd.date_range(uj.index.min(), uj.index.max(), freq='D')).ffill().dropna() uj['spread'] = uj['us2y'] - uj['jp2y'] uj['z_mean'] = uj['spread'].rolling(UJ_Z_WINDOW, min_periods=UJ_Z_WINDOW).mean() uj['z_std'] = uj['spread'].rolling(UJ_Z_WINDOW, min_periods=UJ_Z_WINDOW).std() uj['zscore'] = (uj['spread'] - uj['z_mean']) / uj['z_std'] uj = uj.dropna(subset=['zscore']) cutoff = pd.Timestamp.now() - pd.Timedelta(weeks=8) uj_recent = uj[uj.index >= cutoff] print(f"\n {'Date':>12} {'Spread':>8} {'Z-Score':>8} {'Status':>25}") print(f" {'─'*60}") signals_uj = 0 for date, row in uj_recent.iterrows(): z = row['zscore'] sp = row['spread'] if abs(z) >= UJ_THRESHOLD: status = f"🚨 SIGNAL ({'LONG' if z>0 else 'SHORT'})" signals_uj += 1 elif abs(z) >= 1.5: status = f"⚠️ Approaching ({abs(z):.2f})" else: status = "No signal" marker = " ◄ BREACH" if abs(z) >= UJ_THRESHOLD else "" print(f" {str(date.date()):>12} {sp:>7.3f}% {z:>+8.3f} {status}{marker}") print(f"\n Summary (exact live computation):") print(f" Z-score range: {uj_recent['zscore'].min():.3f} to {uj_recent['zscore'].max():.3f}") print(f" Max |z| reached: {uj_recent['zscore'].abs().max():.3f}") print(f" Threshold breaches (±{UJ_THRESHOLD}): {signals_uj}") except Exception as e: print(f" ❌ USDJPY model failed: {e}") import traceback; traceback.print_exc() print(f"\n{'='*70}") print(" VERDICT") print(f"{'='*70}") print(f""" EURUSD: The beta model z-score can differ significantly from the simple spread z-score. If breaches show here that didn't in v1, compare the dates against your live monitor log to verify trades fired. USDJPY: This IS the exact live computation. Any breaches here that didn't produce live trades need investigating. Key dates to check: search live_monitor_v2.log for lines containing "Z-score" and compare dates against breach dates above. """)