""" signal_timing_analysis_v1.py ============================= Analyses which hours within the session window generate the most z-score threshold crossings and trades. Run from project root on home machine: python src/research/signal_timing_analysis_v1.py """ import pandas as pd import numpy as np import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from pathlib import Path import sys BASE_PATH = Path(__file__).resolve().parents[2] SRC_PATH = BASE_PATH / "src" TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" CHARTS_DIR = BASE_PATH / "data" / "processed" if str(SRC_PATH) not in sys.path: sys.path.append(str(SRC_PATH)) def main(): print("=" * 65) print("SIGNAL TIMING ANALYSIS — Which hours fire the most?") print("=" * 65) # Load trade log trade_log = None for fname in ["trades_real_costs.csv", "trades_eurusd_final.csv"]: p = TRADES_DIR / fname if p.exists(): trade_log = pd.read_csv(p, parse_dates=["entry_time"]) print(f"\n Loaded: {fname} ({len(trade_log):,} trades)") break if trade_log is None: print("ERROR: No trade log found"); return # Load signal series for z-score crossings try: from features.spot_lag_v3 import get_model_ready_spot_lag_v3 signal_df = get_model_ready_spot_lag_v3().copy() signal_df["datetime"] = pd.to_datetime(signal_df["datetime"]) z_col = "lag_zscore_24h_v3" except Exception as e: print(f" Warning: {e}") signal_df = None # ── PART 1: Entry time distribution ────────────────────────────────────── print(f"\n{'─'*65}") print("PART 1: WHAT TIME DO TRADES ACTUALLY ENTER? (UTC)") print(f"{'─'*65}") # EET = UTC+2, session is 07:00-17:00 EET = 05:00-15:00 UTC trade_log["entry_hour_utc"] = trade_log["entry_time"].dt.hour trade_log["entry_hour_eet"] = (trade_log["entry_time"].dt.hour + 2) % 24 hourly = trade_log.groupby("entry_hour_utc").agg( trades = ("entry_time", "count"), win_rate = ("return_real" if "return_real" in trade_log.columns else "return", lambda x: (x > 0).mean() * 100) ).reset_index() total = hourly["trades"].sum() print(f"\n {'UTC Hour':<10}{'EET Hour':<10}{'Trades':>8}{'% of day':>10}" f"{'WR%':>8} Bar") print(f" {'─'*60}") for _, row in hourly.iterrows(): utc_h = int(row["entry_hour_utc"]) eet_h = (utc_h + 2) % 24 n = int(row["trades"]) pct = n / total * 100 wr = row["win_rate"] bar = "█" * int(pct / 1.5) print(f" {utc_h:02d}:00 UTC {eet_h:02d}:00 EET " f"{n:>8,}{pct:>10.1f}%{wr:>8.1f}% {bar}") peak_hour = hourly.loc[hourly["trades"].idxmax(), "entry_hour_utc"] peak_eet = (int(peak_hour) + 2) % 24 print(f"\n Peak entry hour: {int(peak_hour):02d}:00 UTC " f"({peak_eet:02d}:00 EET / " f"{int(peak_hour):02d}:00 London GMT)") # ── PART 2: Z-score crossing times ─────────────────────────────────────── if signal_df is not None: print(f"\n{'─'*65}") print("PART 2: WHAT TIME DOES Z-SCORE CROSS 2.75? (UTC)") print(f" (signal fires, then Fib entry may come later)") print(f"{'─'*65}") THRESHOLD = 2.75 ALLOWED = set(range(5, 15)) # 05:00-14:00 UTC crossings = [] prev_z = 0.0 for _, row in signal_df.iterrows(): z = float(row[z_col]) dt = row["datetime"] if dt.hour not in ALLOWED: prev_z = z; continue if (prev_z < THRESHOLD and z >= THRESHOLD) or \ (prev_z > -THRESHOLD and z <= -THRESHOLD): crossings.append({ "datetime": dt, "hour_utc": dt.hour, "hour_eet": (dt.hour + 2) % 24, "zscore" : z, "direction": "LONG" if z >= THRESHOLD else "SHORT" }) prev_z = z if crossings: cross_df = pd.DataFrame(crossings) cross_hourly = cross_df.groupby("hour_utc").size().reset_index(name="crossings") cross_total = cross_hourly["crossings"].sum() print(f"\n Total threshold crossings: {cross_total:,}") print(f"\n {'UTC Hour':<10}{'EET Hour':<10}{'Crossings':>11}" f"{'% of total':>12} Bar") print(f" {'─'*58}") for _, row in cross_hourly.iterrows(): utc_h = int(row["hour_utc"]) eet_h = (utc_h + 2) % 24 n = int(row["crossings"]) pct = n / cross_total * 100 bar = "█" * int(pct / 1.5) print(f" {utc_h:02d}:00 UTC {eet_h:02d}:00 EET " f"{n:>11,}{pct:>12.1f}% {bar}") peak_cross = cross_hourly.loc[ cross_hourly["crossings"].idxmax(), "hour_utc"] peak_cross_eet = (int(peak_cross) + 2) % 24 print(f"\n Peak crossing hour: {int(peak_cross):02d}:00 UTC " f"({peak_cross_eet:02d}:00 EET / " f"{int(peak_cross):02d}:00 London GMT)") # Direction breakdown dir_counts = cross_df["direction"].value_counts() print(f"\n Long signals : {dir_counts.get('LONG',0):,} " f"({dir_counts.get('LONG',0)/len(cross_df)*100:.1f}%)") print(f" Short signals : {dir_counts.get('SHORT',0):,} " f"({dir_counts.get('SHORT',0)/len(cross_df)*100:.1f}%)") # ── PART 3: Day of week analysis ────────────────────────────────────────── print(f"\n{'─'*65}") print("PART 3: WHICH DAY OF WEEK HAS MOST TRADES?") print(f"{'─'*65}") ret_col = next((c for c in ["return_real","return","ret"] if c in trade_log.columns), None) trade_log["dow"] = trade_log["entry_time"].dt.day_name() dow_order = ["Monday","Tuesday","Wednesday","Thursday","Friday"] dow_stats = trade_log.groupby("dow").agg( trades = ("entry_time","count"), win_rate = (ret_col, lambda x: (x>0).mean()*100) if ret_col else ("entry_time","count") ).reindex(dow_order) dow_total = dow_stats["trades"].sum() print(f"\n {'Day':<12}{'Trades':>8}{'% of week':>11}{'WR%':>8} Bar") print(f" {'─'*55}") for day, row in dow_stats.iterrows(): n = int(row["trades"]) pct = n / dow_total * 100 wr = row["win_rate"] bar = "█" * int(pct / 1.5) print(f" {day:<12}{n:>8,}{pct:>11.1f}%{wr:>8.1f}% {bar}") # ── PART 4: Chart ───────────────────────────────────────────────────────── print(f"\nGenerating chart...") BG = "#0d1117" PANEL = "#161b22" GRID = "#21262d" TEXT = "#e6edf3" MUTED = "#8b949e" GREEN = "#00ff88" RED = "#ff4444" BLUE = "#00d4ff" fig, axes = plt.subplots(1, 3, figsize=(18, 6)) fig.patch.set_facecolor(BG) # Plot 1: Trades by hour ax1 = axes[0] ax1.set_facecolor(PANEL) hours = hourly["entry_hour_utc"].values counts = hourly["trades"].values bars = ax1.bar(hours, counts, color=GREEN, alpha=0.8, width=0.7) ax1.set_xlabel("Hour (UTC)", color=MUTED, fontsize=10) ax1.set_ylabel("Number of trades", color=MUTED, fontsize=10) ax1.set_title("Trade entries by hour (UTC)", color=TEXT, fontsize=11, fontweight="bold") ax1.tick_params(colors=TEXT) for sp in ax1.spines.values(): sp.set_color(GRID) ax1.grid(axis="y", color=GRID, alpha=0.5) ax1.set_xticks(hours) # Plot 2: Z-score crossings by hour ax2 = axes[1] ax2.set_facecolor(PANEL) if signal_df is not None and crossings: c_hours = cross_hourly["hour_utc"].values c_counts = cross_hourly["crossings"].values ax2.bar(c_hours, c_counts, color=BLUE, alpha=0.8, width=0.7) ax2.set_title("Z-score threshold crossings by hour (UTC)", color=TEXT, fontsize=11, fontweight="bold") ax2.set_xlabel("Hour (UTC)", color=MUTED, fontsize=10) ax2.set_ylabel("Threshold crossings", color=MUTED, fontsize=10) ax2.tick_params(colors=TEXT) for sp in ax2.spines.values(): sp.set_color(GRID) ax2.grid(axis="y", color=GRID, alpha=0.5) # Plot 3: Day of week ax3 = axes[2] ax3.set_facecolor(PANEL) days = dow_stats.index.tolist() d_cnts = dow_stats["trades"].values colors = [GREEN if i != dow_stats["trades"].values.argmax() else "#ffd700" for i in range(len(days))] ax3.bar(range(len(days)), d_cnts, color=colors, alpha=0.8, width=0.7) ax3.set_xticks(range(len(days))) ax3.set_xticklabels([d[:3] for d in days], color=TEXT) ax3.set_title("Trades by day of week", color=TEXT, fontsize=11, fontweight="bold") ax3.set_xlabel("Day", color=MUTED, fontsize=10) ax3.set_ylabel("Number of trades", color=MUTED, fontsize=10) ax3.tick_params(colors=TEXT) for sp in ax3.spines.values(): sp.set_color(GRID) ax3.grid(axis="y", color=GRID, alpha=0.5) fig.suptitle("Signal Timing Analysis — When does the model fire?", color=TEXT, fontsize=13, fontweight="bold", y=1.02) plt.tight_layout() chart_path = CHARTS_DIR / "signal_timing_chart.png" plt.savefig(chart_path, dpi=150, bbox_inches="tight", facecolor=fig.get_facecolor()) plt.close() print(f" Chart saved: {chart_path}") print(f"\n Open: {chart_path}") if __name__ == "__main__": main()