Harwinder Sekhon
14448c10e8
EUR/USD never had a validated edge (2026-07-11/12 verdict: no config
passed validation on M5/M15/H1 at 1-pip cost; cost-to-volatility root
cause). Removed: EurUsdStrategyConfig, pair_signals/eurusd.py, EUR
gates 6h + weekend/day filters + strategy marker + trade geometry in
signal_aggregator, EUR pair registry/aliases in main.py and
gpu_backtest.py, EUR/USD news pair mappings, test_all.py Module 10.
Generic multi-currency news keyword infra kept (GBP/JPY/EUR keyword
table is not pair code). Full EUR history remains at 7d3fd95.
Suite: 162/162 green (was 187; 25 EUR-specific tests removed).
Claude-Session: https://claude.ai/code/session_015kby7M9JwpYi8NQT2yoxAi
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""Load and query the historical economic calendar (backtest side).
|
|
|
|
The store is built by `ff_calendar_scraper.py`. Timestamps are unix
|
|
epoch (UTC) — verified against known release times and price-volatility
|
|
alignment (see scraper docstring).
|
|
|
|
Key API:
|
|
cal = load_calendar() # full DataFrame, `utc` column
|
|
ev = precise_events(cal, ["USD"]) # high-impact, minute-accurate
|
|
mask = event_window_mask(bar_index, ev.utc, before, after)
|
|
"""
|
|
from pathlib import Path
|
|
from typing import Iterable, Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
PARQUET = Path(__file__).resolve().parent.parent / "cache" / "calendar.parquet"
|
|
|
|
# timeLabel values that mean "no intraday time" — these events cannot
|
|
# anchor an M5/M15 window (German Prelim CPI is the notable High one).
|
|
_UNTIMED = {"All Day", "Tentative", ""}
|
|
|
|
|
|
def _parse_release_value(raw) -> float:
|
|
"""'0.3%' / '-1.7%' / '228K' / '1.1M' -> float (nan if unparseable)."""
|
|
import re
|
|
if raw is None or (isinstance(raw, float) and np.isnan(raw)):
|
|
return np.nan
|
|
m = re.search(r"(-?\d+(?:\.\d+)?)\s*([kmbt])?",
|
|
str(raw).replace(",", ""), re.IGNORECASE)
|
|
if not m:
|
|
return np.nan
|
|
mult = {"k": 1e3, "m": 1e6, "b": 1e9, "t": 1e12}.get(
|
|
(m.group(2) or "").lower(), 1.0)
|
|
return float(m.group(1)) * mult
|
|
|
|
|
|
def load_calendar(path: Path = PARQUET) -> pd.DataFrame:
|
|
df = pd.read_parquet(path)
|
|
df["utc"] = pd.to_datetime(df.dateline, unit="s", utc=True)
|
|
for col in ("actual", "forecast", "previous"):
|
|
df[f"{col}_num"] = df[col].map(_parse_release_value)
|
|
df["surprise"] = df.actual_num - df.forecast_num
|
|
return df
|
|
|
|
|
|
def precise_events(
|
|
cal: pd.DataFrame,
|
|
currencies: Iterable[str] = ("USD",),
|
|
impacts: Iterable[str] = ("high",),
|
|
require_values: bool = False,
|
|
) -> pd.DataFrame:
|
|
"""Events with a minute-accurate timestamp, filtered by currency/impact."""
|
|
out = cal[cal.currency.isin(list(currencies))
|
|
& cal.impactName.isin(list(impacts))
|
|
& ~cal.timeMasked.astype(bool)
|
|
& ~cal.timeLabel.fillna("").isin(_UNTIMED)]
|
|
if require_values:
|
|
out = out.dropna(subset=["actual_num", "forecast_num"])
|
|
return out.sort_values("utc").reset_index(drop=True)
|
|
|
|
|
|
def event_window_mask(
|
|
bar_index: pd.DatetimeIndex,
|
|
event_times: pd.Series,
|
|
before: pd.Timedelta = pd.Timedelta(minutes=30),
|
|
after: pd.Timedelta = pd.Timedelta(minutes=30),
|
|
) -> np.ndarray:
|
|
"""Boolean mask: bar OPEN time falls inside [event-before, event+after).
|
|
|
|
Use on a backtest's entry times to identify trades entered around
|
|
releases (avoidance counterfactuals, event-window cost models).
|
|
"""
|
|
starts = pd.DatetimeIndex(event_times) - before
|
|
ends = pd.DatetimeIndex(event_times) + after
|
|
order = np.argsort(starts.values)
|
|
starts, ends = starts.values[order], ends.values[order]
|
|
# merge overlapping windows, then interval-stab each bar time
|
|
merged_s, merged_e = [], []
|
|
for s, e in zip(starts, ends):
|
|
if merged_e and s <= merged_e[-1]:
|
|
merged_e[-1] = max(merged_e[-1], e)
|
|
else:
|
|
merged_s.append(s)
|
|
merged_e.append(e)
|
|
ms = np.array(merged_s)
|
|
me = np.array(merged_e)
|
|
ts = bar_index.values
|
|
pos = np.searchsorted(ms, ts, side="right") - 1
|
|
ok = pos >= 0
|
|
res = np.zeros(len(ts), dtype=bool)
|
|
res[ok] = ts[ok] < me[pos[ok]]
|
|
return res
|