305 lines
12 KiB
Python
305 lines
12 KiB
Python
|
"""
|
|||
|
|
analysis/time_performance.py
|
|||
|
|
Analyzes trade performance by hour (UTC), session, and day of week.
|
|||
|
|
Identifies statistically significant negative-edge time windows.
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import numpy as np
|
|||
|
|
import pandas as pd
|
|||
|
|
|
|||
|
|
from data.models import Finding, RunMetrics
|
|||
|
|
from analysis.base import BaseAnalyzer
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TimePerformanceAnalyzer(BaseAnalyzer):
|
|||
|
|
"""
|
|||
|
|
Session / Hour / Day-of-Week Performance Analyzer.
|
|||
|
|
|
|||
|
|
Buckets trades by time dimension and flags windows with:
|
|||
|
|
- Z-score < threshold (mean PnL very negative vs overall)
|
|||
|
|
- Statistically significant by permutation test (p < 0.10)
|
|||
|
|
- Minimum trade count (don't flag buckets with too few trades)
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
name = "time_performance"
|
|||
|
|
|
|||
|
|
def __init__(
|
|||
|
|
self,
|
|||
|
|
z_score_threshold: float = -1.5,
|
|||
|
|
min_bucket_trades: int = 10,
|
|||
|
|
permutation_n: int = 1000,
|
|||
|
|
pvalue_threshold: float = 0.10,
|
|||
|
|
):
|
|||
|
|
self.z_threshold = z_score_threshold
|
|||
|
|
self.min_bucket = min_bucket_trades
|
|||
|
|
self.perm_n = permutation_n
|
|||
|
|
self.p_threshold = pvalue_threshold
|
|||
|
|
|
|||
|
|
def analyze(self, trades: pd.DataFrame, metrics: RunMetrics) -> list[Finding]:
|
|||
|
|
if "hour_utc" not in trades.columns:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
findings = []
|
|||
|
|
findings += self._analyze_hours(trades, metrics)
|
|||
|
|
findings += self._analyze_sessions(trades, metrics)
|
|||
|
|
findings += self._analyze_days(trades, metrics)
|
|||
|
|
return sorted(findings, key=lambda f: f.confidence, reverse=True)
|
|||
|
|
|
|||
|
|
# ── Hour analysis ─────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
def _analyze_hours(self, df: pd.DataFrame, metrics: RunMetrics) -> list[Finding]:
|
|||
|
|
"""Flag individual UTC hours with poor performance."""
|
|||
|
|
global_mean = df["net_money"].mean()
|
|||
|
|
global_std = df["net_money"].std()
|
|||
|
|
all_pnl = df["net_money"].values
|
|||
|
|
|
|||
|
|
if global_std == 0:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
findings = []
|
|||
|
|
for hour in sorted(df["hour_utc"].dropna().unique()):
|
|||
|
|
bucket = df[df["hour_utc"] == hour]
|
|||
|
|
if len(bucket) < self.min_bucket:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
mean_pnl = bucket["net_money"].mean()
|
|||
|
|
z = (mean_pnl - global_mean) / global_std
|
|||
|
|
|
|||
|
|
if z >= self.z_threshold:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
# Permutation test
|
|||
|
|
p_val = self._permutation_pvalue(
|
|||
|
|
bucket["net_money"].values, all_pnl,
|
|||
|
|
n_permutations=self.perm_n, alternative="less"
|
|||
|
|
)
|
|||
|
|
if p_val >= self.p_threshold:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
impact_pnl = abs(float(bucket[bucket["net_money"] < 0]["net_money"].sum()))
|
|||
|
|
confidence = max(0.0, min(0.97, 1 - p_val))
|
|||
|
|
|
|||
|
|
findings.append(Finding(
|
|||
|
|
run_id=self.run_id,
|
|||
|
|
analyzer=self.name,
|
|||
|
|
description=(
|
|||
|
|
f"Hour {hour:02d}:00 UTC: mean PnL ${mean_pnl:.2f} "
|
|||
|
|
f"(Z={z:.2f}, {len(bucket)} trades). "
|
|||
|
|
f"Estimated negative contribution: ${impact_pnl:.0f}."
|
|||
|
|
),
|
|||
|
|
severity=self._severity(confidence, impact_pnl, metrics.net_profit),
|
|||
|
|
confidence=confidence,
|
|||
|
|
impact_estimate_pnl=impact_pnl,
|
|||
|
|
suggested_params={}, # session filter suggestion built by aggregator
|
|||
|
|
evidence={
|
|||
|
|
"type": "hour",
|
|||
|
|
"hour_utc": int(hour),
|
|||
|
|
"mean_pnl": round(mean_pnl, 2),
|
|||
|
|
"z_score": round(z, 3),
|
|||
|
|
"trade_count": int(len(bucket)),
|
|||
|
|
"p_value": round(p_val, 4),
|
|||
|
|
},
|
|||
|
|
))
|
|||
|
|
|
|||
|
|
# Consolidate consecutive bad hours into a single window finding
|
|||
|
|
if findings:
|
|||
|
|
findings = self._consolidate_hour_findings(findings, df, metrics)
|
|||
|
|
|
|||
|
|
return findings
|
|||
|
|
|
|||
|
|
def _consolidate_hour_findings(
|
|||
|
|
self, hour_findings: list[Finding], df: pd.DataFrame, metrics: RunMetrics
|
|||
|
|
) -> list[Finding]:
|
|||
|
|
"""
|
|||
|
|
Group consecutive flagged hours into a single window finding.
|
|||
|
|
E.g. hours [14, 15, 16] → "14:00–17:00 UTC bad window"
|
|||
|
|
Returns a single consolidated finding (plus keeps top individual for detail).
|
|||
|
|
"""
|
|||
|
|
bad_hours = sorted(
|
|||
|
|
int(f.evidence["hour_utc"]) for f in hour_findings
|
|||
|
|
)
|
|||
|
|
if not bad_hours:
|
|||
|
|
return hour_findings
|
|||
|
|
|
|||
|
|
# Find contiguous groups
|
|||
|
|
groups = []
|
|||
|
|
group = [bad_hours[0]]
|
|||
|
|
for h in bad_hours[1:]:
|
|||
|
|
if h == group[-1] + 1:
|
|||
|
|
group.append(h)
|
|||
|
|
else:
|
|||
|
|
groups.append(group)
|
|||
|
|
group = [h]
|
|||
|
|
groups.append(group)
|
|||
|
|
|
|||
|
|
consolidated = []
|
|||
|
|
for g in groups:
|
|||
|
|
start_h = g[0]
|
|||
|
|
end_h = g[-1] + 1
|
|||
|
|
window = df[df["hour_utc"].between(start_h, g[-1])]
|
|||
|
|
total_pnl = float(window["net_money"].sum())
|
|||
|
|
n_trades = len(window)
|
|||
|
|
impact = abs(float(window[window["net_money"] < 0]["net_money"].sum()))
|
|||
|
|
|
|||
|
|
# Derive session filter params from window
|
|||
|
|
# Convert UTC to broker local time for session params
|
|||
|
|
broker_start = (start_h + 2) % 24 # UTC+2 (from config)
|
|||
|
|
broker_end = (end_h + 2) % 24
|
|||
|
|
|
|||
|
|
max_conf = max(f.confidence for f in hour_findings
|
|||
|
|
if f.evidence["hour_utc"] in g)
|
|||
|
|
|
|||
|
|
consolidated.append(Finding(
|
|||
|
|
run_id=self.run_id,
|
|||
|
|
analyzer=self.name,
|
|||
|
|
description=(
|
|||
|
|
f"Negative edge window {start_h:02d}:00–{end_h:02d}:00 UTC: "
|
|||
|
|
f"${total_pnl:.0f} total, {n_trades} trades. "
|
|||
|
|
f"Consider excluding this window via session filter."
|
|||
|
|
),
|
|||
|
|
severity=self._severity(max_conf, impact, metrics.net_profit),
|
|||
|
|
confidence=max_conf,
|
|||
|
|
impact_estimate_pnl=impact,
|
|||
|
|
suggested_params={
|
|||
|
|
"InpUseSession": True,
|
|||
|
|
# Preserve existing session start; cut end before bad window
|
|||
|
|
# These are broker-local hours
|
|||
|
|
"InpSessionEnd": (broker_start) % 24,
|
|||
|
|
},
|
|||
|
|
evidence={
|
|||
|
|
"type": "hour_window",
|
|||
|
|
"start_utc": start_h,
|
|||
|
|
"end_utc": end_h,
|
|||
|
|
"broker_start": broker_start,
|
|||
|
|
"broker_end": broker_end,
|
|||
|
|
"total_pnl": round(total_pnl, 2),
|
|||
|
|
"trade_count": n_trades,
|
|||
|
|
"hours_flagged": g,
|
|||
|
|
},
|
|||
|
|
))
|
|||
|
|
|
|||
|
|
return consolidated
|
|||
|
|
|
|||
|
|
# ── Session analysis ──────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
def _analyze_sessions(self, df: pd.DataFrame, metrics: RunMetrics) -> list[Finding]:
|
|||
|
|
if "session" not in df.columns:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
global_mean = df["net_money"].mean()
|
|||
|
|
global_std = df["net_money"].std()
|
|||
|
|
all_pnl = df["net_money"].values
|
|||
|
|
|
|||
|
|
if global_std == 0:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
findings = []
|
|||
|
|
for session in df["session"].dropna().unique():
|
|||
|
|
bucket = df[df["session"] == session]
|
|||
|
|
if len(bucket) < self.min_bucket:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
mean_pnl = bucket["net_money"].mean()
|
|||
|
|
z = (mean_pnl - global_mean) / global_std
|
|||
|
|
if z >= self.z_threshold:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
p_val = self._permutation_pvalue(
|
|||
|
|
bucket["net_money"].values, all_pnl,
|
|||
|
|
n_permutations=self.perm_n, alternative="less"
|
|||
|
|
)
|
|||
|
|
if p_val >= self.p_threshold:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
pf = (
|
|||
|
|
bucket[bucket["net_money"] > 0]["net_money"].sum() /
|
|||
|
|
max(0.01, abs(bucket[bucket["net_money"] < 0]["net_money"].sum()))
|
|||
|
|
)
|
|||
|
|
impact = abs(float(bucket[bucket["net_money"] < 0]["net_money"].sum()))
|
|||
|
|
confidence = max(0.0, min(0.97, 1 - p_val))
|
|||
|
|
|
|||
|
|
findings.append(Finding(
|
|||
|
|
run_id=self.run_id,
|
|||
|
|
analyzer=self.name,
|
|||
|
|
description=(
|
|||
|
|
f"{session} session: PF {pf:.2f}, mean PnL ${mean_pnl:.2f} "
|
|||
|
|
f"(Z={z:.2f}, {len(bucket)} trades). Recommend excluding this session."
|
|||
|
|
),
|
|||
|
|
severity=self._severity(confidence, impact, metrics.net_profit),
|
|||
|
|
confidence=confidence,
|
|||
|
|
impact_estimate_pnl=impact,
|
|||
|
|
suggested_params={"InpUseSession": True},
|
|||
|
|
evidence={
|
|||
|
|
"type": "session",
|
|||
|
|
"session": session,
|
|||
|
|
"profit_factor": round(float(pf), 3),
|
|||
|
|
"mean_pnl": round(mean_pnl, 2),
|
|||
|
|
"z_score": round(z, 3),
|
|||
|
|
"trade_count": int(len(bucket)),
|
|||
|
|
"p_value": round(p_val, 4),
|
|||
|
|
},
|
|||
|
|
))
|
|||
|
|
|
|||
|
|
return findings
|
|||
|
|
|
|||
|
|
# ── Day-of-week analysis ──────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
def _analyze_days(self, df: pd.DataFrame, metrics: RunMetrics) -> list[Finding]:
|
|||
|
|
if "day_of_week" not in df.columns:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"]
|
|||
|
|
global_mean = df["net_money"].mean()
|
|||
|
|
global_std = df["net_money"].std()
|
|||
|
|
all_pnl = df["net_money"].values
|
|||
|
|
|
|||
|
|
if global_std == 0:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
findings = []
|
|||
|
|
for day in range(5): # 0=Mon … 4=Fri
|
|||
|
|
bucket = df[df["day_of_week"] == day]
|
|||
|
|
if len(bucket) < self.min_bucket:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
mean_pnl = bucket["net_money"].mean()
|
|||
|
|
z = (mean_pnl - global_mean) / global_std
|
|||
|
|
if z >= self.z_threshold:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
p_val = self._permutation_pvalue(
|
|||
|
|
bucket["net_money"].values, all_pnl,
|
|||
|
|
n_permutations=self.perm_n, alternative="less"
|
|||
|
|
)
|
|||
|
|
if p_val >= self.p_threshold:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
impact = abs(float(bucket[bucket["net_money"] < 0]["net_money"].sum()))
|
|||
|
|
confidence = max(0.0, min(0.97, 1 - p_val))
|
|||
|
|
|
|||
|
|
findings.append(Finding(
|
|||
|
|
run_id=self.run_id,
|
|||
|
|
analyzer=self.name,
|
|||
|
|
description=(
|
|||
|
|
f"{DAY_NAMES[day]}: mean PnL ${mean_pnl:.2f} "
|
|||
|
|
f"(Z={z:.2f}, {len(bucket)} trades). "
|
|||
|
|
f"Possible day-of-week edge degradation."
|
|||
|
|
),
|
|||
|
|
severity="low", # day-level findings are informational
|
|||
|
|
confidence=confidence,
|
|||
|
|
impact_estimate_pnl=impact,
|
|||
|
|
suggested_params={},
|
|||
|
|
evidence={
|
|||
|
|
"type": "day_of_week",
|
|||
|
|
"day": DAY_NAMES[day],
|
|||
|
|
"day_index": day,
|
|||
|
|
"mean_pnl": round(mean_pnl, 2),
|
|||
|
|
"z_score": round(z, 3),
|
|||
|
|
"trade_count": int(len(bucket)),
|
|||
|
|
"p_value": round(p_val, 4),
|
|||
|
|
},
|
|||
|
|
))
|
|||
|
|
|
|||
|
|
return findings
|