Add calendar research infra: FF scraper + verified event store
Scrapes Forex Factory weekly pages (epoch datelines, actual/forecast/ impact/betterWorse) into backtest/cache/calendar.parquet (gitignored; rebuild with the scraper). Timestamps verified: NFP/CPI/ECB times exact across DST, and high-impact event bars average 3.5x ATR with the spike vanishing at +/-1h shifts. calendar_store provides precise_events and event_window_mask for screens and counterfactuals.
This commit is contained in:
Harwinder Sekhon
@@ -0,0 +1,5 @@
|
||||
"""Research tooling: calendar sourcing/store and edge-screening utilities.
|
||||
|
||||
Nothing here is imported by the live bot; these modules support
|
||||
backtesting and edge validation only.
|
||||
"""
|
||||
@@ -0,0 +1,94 @@
|
||||
"""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", "EUR"]) # 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", "EUR"),
|
||||
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
|
||||
@@ -0,0 +1,144 @@
|
||||
"""Forex Factory calendar scraper -> backtest/cache/calendar.parquet.
|
||||
|
||||
Fetches weekly calendar pages and parses the event JSON embedded in
|
||||
`window.calendarComponentStates` (the `days:` array). Event `dateline`
|
||||
is a unix epoch, so timestamps carry no timezone ambiguity — verified
|
||||
2026-07-11 against known release times (NFP 12:30/13:30 UTC across DST,
|
||||
ECB 12:15/13:15) and against EUR/USD M15 volatility: the bar containing
|
||||
a high-impact event averages 3.5x normal range, and shifting event
|
||||
times by ±1h destroys that alignment.
|
||||
|
||||
Usage:
|
||||
python backtest/research/ff_calendar_scraper.py --start 2024-01-01 --end 2026-07-13
|
||||
python backtest/research/ff_calendar_scraper.py # refresh: last stored week -> today
|
||||
|
||||
Merges into the existing parquet on event (id, dateline), so refreshing
|
||||
is idempotent and also picks up revised `actual` values for recent weeks.
|
||||
Be polite: ~1.6s between requests (a full year is ~52 requests).
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import time
|
||||
from datetime import date, timedelta
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
import pandas as pd
|
||||
import requests
|
||||
|
||||
CACHE_DIR = Path(__file__).resolve().parent.parent / "cache"
|
||||
PARQUET = CACHE_DIR / "calendar.parquet"
|
||||
|
||||
UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/126.0 Safari/537.36")
|
||||
|
||||
FIELDS = ["id", "dateline", "name", "country", "currency", "impactName",
|
||||
"timeLabel", "timeMasked", "actual", "forecast", "previous",
|
||||
"revision", "actualBetterWorse"]
|
||||
|
||||
|
||||
def week_url(d: date) -> str:
|
||||
return (f"https://www.forexfactory.com/calendar?week="
|
||||
f"{d.strftime('%b').lower()}{d.day}.{d.year}")
|
||||
|
||||
|
||||
def extract_days(html: str) -> Optional[list]:
|
||||
"""Balanced-bracket extraction of the `days:` JSON array."""
|
||||
i = html.find("days:")
|
||||
if i < 0:
|
||||
return None
|
||||
j = html.find("[", i)
|
||||
depth, in_str, esc = 0, False, False
|
||||
for k in range(j, len(html)):
|
||||
ch = html[k]
|
||||
if in_str:
|
||||
if esc:
|
||||
esc = False
|
||||
elif ch == "\\":
|
||||
esc = True
|
||||
elif ch == '"':
|
||||
in_str = False
|
||||
else:
|
||||
if ch == '"':
|
||||
in_str = True
|
||||
elif ch == "[":
|
||||
depth += 1
|
||||
elif ch == "]":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return json.loads(html[j:k + 1])
|
||||
return None
|
||||
|
||||
|
||||
def fetch_week(sess: requests.Session, d: date, retries: int = 4) -> Optional[List[dict]]:
|
||||
for attempt in range(retries):
|
||||
try:
|
||||
r = sess.get(week_url(d), timeout=30)
|
||||
if r.status_code == 200:
|
||||
days = extract_days(r.text)
|
||||
if days is not None:
|
||||
return [{k: e.get(k) for k in FIELDS}
|
||||
for day in days for e in day.get("events", [])]
|
||||
except requests.RequestException:
|
||||
pass
|
||||
time.sleep(5 * (attempt + 1))
|
||||
return None
|
||||
|
||||
|
||||
def scrape(start: date, end: date) -> pd.DataFrame:
|
||||
sess = requests.Session()
|
||||
sess.headers["User-Agent"] = UA
|
||||
rows: List[dict] = []
|
||||
failed: List[date] = []
|
||||
d = start
|
||||
while d <= end:
|
||||
week_rows = fetch_week(sess, d)
|
||||
if week_rows is None:
|
||||
failed.append(d)
|
||||
print(f"FAILED week {d}")
|
||||
else:
|
||||
rows.extend(week_rows)
|
||||
d += timedelta(days=7)
|
||||
time.sleep(1.6)
|
||||
if failed:
|
||||
print(f"WARNING: {len(failed)} weeks failed: {failed}")
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
def merge_into_store(new: pd.DataFrame, path: Path = PARQUET) -> pd.DataFrame:
|
||||
if path.exists():
|
||||
old = pd.read_parquet(path)
|
||||
combined = pd.concat([old, new])
|
||||
else:
|
||||
combined = new
|
||||
# keep="last": a refresh overwrites stale rows (actuals fill in
|
||||
# after release, revisions land weeks later).
|
||||
combined = (combined.drop_duplicates(subset=["id", "dateline"], keep="last")
|
||||
.sort_values("dateline").reset_index(drop=True))
|
||||
combined.to_parquet(path, index=False)
|
||||
return combined
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
ap.add_argument("--start", type=date.fromisoformat, default=None)
|
||||
ap.add_argument("--end", type=date.fromisoformat, default=date.today())
|
||||
args = ap.parse_args()
|
||||
|
||||
start = args.start
|
||||
if start is None:
|
||||
if PARQUET.exists():
|
||||
last = pd.read_parquet(PARQUET).dateline.max()
|
||||
start = pd.Timestamp(last, unit="s").date() - timedelta(days=7)
|
||||
else:
|
||||
start = date(2024, 1, 1)
|
||||
print(f"Scraping {start} -> {args.end}")
|
||||
new = scrape(start, args.end)
|
||||
combined = merge_into_store(new)
|
||||
print(f"Store: {len(combined)} events "
|
||||
f"({pd.Timestamp(combined.dateline.min(), unit='s')} -> "
|
||||
f"{pd.Timestamp(combined.dateline.max(), unit='s')}) -> {PARQUET}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user