Harwinder Sekhon
af06685542
Adds an opt-in recorder to the backtester so a replay keeps the evidence each trade setup was decided on: every indicator's vote, value, confidence, base weight, regime multiplier, effective weight and contribution on the primary and both confirmation timeframes, penalties, the setup and outcome, and the Forex Factory calendar events around the trade (4 h before entry to exit, all-day rows on the New-York date). --record-bars adds one float32 row per analyzed bar with the same votes plus 1 h/4 h/24 h forward moves in ATRs. Output is Parquet in a folder per run with a manifest carrying the engine/gate fingerprints, the regime weight table in force and the book MD5. Nothing in the signal core changes: the trade book is byte-identical with recording on or off (control-vs-recorded MD5 equal on both 2-year runs, M5 781 / +146.07R / PF 1.52, M15 793 / +117.62R / PF 1.41 at $0.65). Module 33 adds 12 tests (498 total): snapshot reproduces the engine's composite, absence is no vote, calendar join edges/phases/holiday, write/load round-trip, _replay_span parity with recording on, CLI flags, import and source guards. Also: backtest/replay_scorecard.py (first-look scorecards, calendar buckets, in-book and per-bar what-ifs, --check invariants), scripts/ff_calendar_scraper.py restored from git history (the only way to regenerate the gitignored calendar file), report in backtest/research/replay_recording_2026-09-09.md. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Nc9Ninzyi18QkWezrxeBvB
1104 lines
47 KiB
Python
1104 lines
47 KiB
Python
"""
|
|
ForexBot2 — Backtest decision recorder (RESEARCH, opt-in)
|
|
=========================================================
|
|
Turns a `--record DIR` replay into a small parquet dataset describing WHY
|
|
every accepted trade was taken: the composite maths of the primary AND
|
|
both confirmation timeframes, every indicator's value / signal / base
|
|
weight / regime multiplier / effective weight / weighted score, the
|
|
penalty tags, the setup, and the economic-calendar events around the
|
|
entry. With `--record-bars` it also stores one compact float32 row per
|
|
ANALYZED primary bar so a study can compare taken trades against the bars
|
|
that produced nothing.
|
|
|
|
Hard contract
|
|
-------------
|
|
* **Read-only w.r.t. the signal path.** Nothing here is imported by
|
|
`signal_aggregator`, `engine` or the live surfaces, and nothing here
|
|
writes back onto a report/setup. The trade book (and therefore
|
|
`--export-csv`) is byte-identical with recording on or off.
|
|
* **Plain data only** crosses the worker boundary: `snapshot_report`
|
|
returns str/int/float/bool/None/list/dict, so pickling a `Trade` back
|
|
from a replay worker costs no enums, timestamps or dataclasses.
|
|
* Allowed imports are stdlib + numpy/pandas + `config.Timeframe` +
|
|
`engine.regime_weight_table`. Never signal_log / news_engine /
|
|
providers / gpu_backtest (circular) / sklearn.
|
|
|
|
The composite arithmetic mirrored below is `engine.TechnicalEngine.
|
|
_compute_composite`; `composite_from_rows` exists so a study can
|
|
reproduce a composite from the stored rows and prove nothing was lost.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import math
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable, List, Optional, Tuple
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
# Ensure project root on sys.path so this module works standalone, as
|
|
# `backtest.replay_record`, and when `backtest/` itself is on sys.path.
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from config import Timeframe # noqa: E402
|
|
from engine import regime_weight_table # noqa: E402
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
# CONSTANTS
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
|
|
INDICATORS = (
|
|
"ema_trend", "rsi", "macd", "bollinger", "adx",
|
|
"roc", "keltner", "divergence", "candlestick",
|
|
)
|
|
ROLES = ("primary", "c1", "c2")
|
|
|
|
# Mirrors engine.TechnicalEngine._compute_composite (engine.py ~1117):
|
|
# score = clamp((Σ wscore / Σ eff_weight) * 50, -100, 100)
|
|
# >= 40 STRONG_BUY | >= 15 BUY | <= -40 STRONG_SELL | <= -15 SELL | else NEUTRAL
|
|
COMPOSITE_THRESHOLDS = {"strong": 40.0, "weak": 15.0}
|
|
|
|
CALENDAR_PARQUET = Path(__file__).resolve().parent / "cache" / "calendar.parquet"
|
|
EVENT_WINDOW = pd.Timedelta(hours=4)
|
|
NEAR_RELEASE_MAX_MIN = 1440.0
|
|
_NY = ZoneInfo("America/New_York")
|
|
|
|
REGIME_CODES = {
|
|
"unknown": 0, "trending_up": 1, "trending_down": 2,
|
|
"trending": 3, "transitioning": 4, "ranging": 5,
|
|
}
|
|
# engine._compute_atr emits exactly these four.
|
|
VOL_CODES = {"low": 0, "normal": 1, "high": 2, "extreme": 3}
|
|
QUALITY_CODES = {"": -1, "F": 0, "D": 1, "C": 2, "B": 3, "A": 4, "A+": 5}
|
|
# "XAU" is the backtester's book tag for the ATR archetype; the live
|
|
# TradeSetup.strategy for the same generator is "XAU_ATR" — both map to 2
|
|
# so `strategy_code` never collides with 0 ("no setup"). An unrecognised
|
|
# strategy stays NaN rather than masquerading as no-setup.
|
|
STRATEGY_CODES = {"": 0, "GENERIC": 1, "XAU": 2, "XAU_ATR": 2,
|
|
"XAU_PB": 3, "XAU_CP": 4}
|
|
|
|
|
|
def _build_bar_cols() -> Tuple[str, ...]:
|
|
"""The float32 vector layout of one recorded bar (10 + 15 + 135 = 160)."""
|
|
cols: List[str] = [
|
|
"close", "final_score", "final_conf", "quality_code", "strategy_code",
|
|
"has_setup", "final_signal", "mtf_agreement", "penalty_total",
|
|
"exhaustion_count",
|
|
]
|
|
for role in ROLES:
|
|
cols += [f"{role}_regime_code", f"{role}_vol_code", f"{role}_composite",
|
|
f"{role}_confidence", f"{role}_atr"]
|
|
for role in ROLES:
|
|
for ind in INDICATORS:
|
|
cols += [f"{role}_{ind}_signal", f"{role}_{ind}_value",
|
|
f"{role}_{ind}_conf", f"{role}_{ind}_eff_weight",
|
|
f"{role}_{ind}_wscore"]
|
|
return tuple(cols)
|
|
|
|
|
|
BAR_COLS = _build_bar_cols()
|
|
_BAR_IDX = {name: i for i, name in enumerate(BAR_COLS)}
|
|
_N_BAR_COLS = len(BAR_COLS)
|
|
|
|
# Every column below is normalised to datetime64[ns, UTC] before writing.
|
|
_DT_COLS = ("bar_ts", "decision_time", "entry_time", "exit_time", "ts_utc")
|
|
|
|
_EVENT_COLS = [
|
|
"trade_id", "event_id", "ts_utc", "name", "currency", "impact", "timed",
|
|
"actual", "forecast", "previous", "revision", "better_worse",
|
|
"offset_min", "phase",
|
|
]
|
|
_IND_COLS = [
|
|
"trade_id", "tf_role", "timeframe", "indicator", "signal", "signal_name",
|
|
"value", "confidence", "base_weight", "regime_mult", "eff_weight",
|
|
"wscore", "agrees", "regime", "description",
|
|
]
|
|
_SCALAR_COLS = [
|
|
"ev_prev_hi_usd_min", "ev_prev_hi_usd_name",
|
|
"ev_next_hi_usd_min", "ev_next_hi_usd_name",
|
|
"ev_hi_usd_during_hold", "ev_hi_usd_during_hold_names",
|
|
"ev_any_hi_during_hold", "ev_entry_within_30m_hi_usd", "us_holiday",
|
|
]
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
# SMALL HELPERS
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
|
|
def git_sha() -> str:
|
|
"""Short HEAD sha of the repo, or "" when git is unavailable."""
|
|
try:
|
|
out = subprocess.run(
|
|
["git", "rev-parse", "--short", "HEAD"],
|
|
cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=2,
|
|
)
|
|
return out.stdout.strip() if out.returncode == 0 else ""
|
|
except Exception: # pragma: no cover
|
|
return ""
|
|
|
|
|
|
def _finite(x) -> Optional[float]:
|
|
"""float(x) when it is a real finite number, else None."""
|
|
if x is None:
|
|
return None
|
|
try:
|
|
v = float(x)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if math.isnan(v) or math.isinf(v):
|
|
return None
|
|
return v
|
|
|
|
|
|
def _num(x) -> float:
|
|
"""_finite with NaN instead of None (for float32 buffers)."""
|
|
v = _finite(x)
|
|
return np.nan if v is None else v
|
|
|
|
|
|
def _agrees(signal_int: int, direction: str) -> str:
|
|
"""Same rule as gpu_backtest._build_indicator_breakdown."""
|
|
is_buy = direction == "BUY"
|
|
if signal_int > 0: # BUY / STRONG_BUY
|
|
return "agree" if is_buy else "oppose"
|
|
if signal_int < 0: # SELL / STRONG_SELL
|
|
return "oppose" if is_buy else "agree"
|
|
return "neutral"
|
|
|
|
|
|
def _sig_int(signal) -> int:
|
|
return int(getattr(signal, "value", 0) or 0)
|
|
|
|
|
|
def _sanitize(name: str) -> str:
|
|
return "".join(ch if (ch.isalnum() or ch == "_") else "_" for ch in str(name))
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
# DECISION SNAPSHOT (runs inside the replay worker)
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
|
|
def _tf_block(analysis, role: str, direction: str) -> Dict:
|
|
"""One timeframe's plain-data block (scored indicators only).
|
|
|
|
Indicators are skipped under exactly the composite's rule — private
|
|
`_`-prefixed series and weight-0 context indicators (that is what
|
|
excludes `atr`) — so Σ eff_weight / Σ wscore over the stored rows
|
|
reproduces the composite.
|
|
"""
|
|
inds: Dict[str, Dict] = {}
|
|
for name, ind in (getattr(analysis, "indicators", None) or {}).items():
|
|
if name.startswith("_") or ind.weight == 0:
|
|
continue
|
|
sig_int = _sig_int(ind.signal)
|
|
base = _finite(ind.weight)
|
|
eff = _finite(getattr(ind, "effective_weight", None))
|
|
regime_mult = (eff / base) if (eff is not None and base) and base > 0 else None
|
|
inds[name] = {
|
|
"signal": sig_int,
|
|
"signal_name": str(getattr(ind.signal, "name", "")),
|
|
"value": _finite(ind.value),
|
|
"confidence": _finite(ind.confidence),
|
|
"base_weight": base,
|
|
"regime_mult": regime_mult,
|
|
"eff_weight": eff,
|
|
"wscore": _finite(getattr(ind, "weighted_score", None)),
|
|
"agrees": _agrees(sig_int, direction),
|
|
"description": str(ind.description or ""),
|
|
}
|
|
levels = getattr(analysis, "key_levels", None) or {}
|
|
return {
|
|
"tf": str(getattr(analysis, "timeframe", "") or ""),
|
|
"role": role,
|
|
"regime": str(getattr(analysis, "market_regime", "") or ""),
|
|
"vol": str(getattr(analysis, "volatility_regime", "") or ""),
|
|
"composite_score": _finite(getattr(analysis, "composite_score", None)),
|
|
"composite_signal": _sig_int(getattr(analysis, "composite_signal", None)),
|
|
"confidence": _finite(getattr(analysis, "confidence", None)),
|
|
"atr": _finite(levels.get("atr")),
|
|
"atr_pct": _finite(levels.get("atr_pct")),
|
|
"pattern_signals": [str(p) for p in (getattr(analysis, "pattern_signals", None) or [])],
|
|
"indicators": inds,
|
|
}
|
|
|
|
|
|
def _role_analyses(report, confirmation_tfs) -> List[Tuple[str, object]]:
|
|
"""[(role, analysis)] — primary first, then c1/c2 by confirmation order."""
|
|
out: List[Tuple[str, object]] = []
|
|
primary = getattr(report, "primary_analysis", None)
|
|
if primary is not None:
|
|
out.append(("primary", primary))
|
|
order = [tf.value for tf in (confirmation_tfs or [])]
|
|
for analysis in (getattr(report, "confirmation_analyses", None) or []):
|
|
tf_name = str(getattr(analysis, "timeframe", "") or "")
|
|
try:
|
|
i = order.index(tf_name)
|
|
except ValueError:
|
|
logger.debug("record: confirmation analysis %s is not in %s — skipped",
|
|
tf_name, order)
|
|
continue
|
|
if i + 1 >= len(ROLES):
|
|
logger.debug("record: more confirmation TFs than roles (%s)", tf_name)
|
|
continue
|
|
out.append((ROLES[i + 1], analysis))
|
|
return out
|
|
|
|
|
|
def snapshot_report(report, setup, decision_time,
|
|
primary_tf, confirmation_tfs) -> Dict:
|
|
"""Plain-data decision snapshot; `json.dumps` must succeed on it.
|
|
|
|
Called inside the replay worker for every accepted candidate, so it
|
|
must stay allocation-light and must not touch the report.
|
|
"""
|
|
direction = str(getattr(setup, "direction", "") or "")
|
|
primary = getattr(report, "primary_analysis", None)
|
|
# report.timestamp is wall-clock in analyze_from_dataframes; the
|
|
# primary analysis carries the bar LABEL (same string signal_log
|
|
# stores as bar_ts / indicators_json).
|
|
bar_ts = str(getattr(primary, "timestamp", "") or "") or str(report.timestamp)
|
|
|
|
try:
|
|
decision_iso = pd.Timestamp(decision_time).isoformat()
|
|
except Exception: # pragma: no cover
|
|
decision_iso = str(decision_time)
|
|
|
|
return {
|
|
"bar_ts": bar_ts,
|
|
"decision_time": decision_iso,
|
|
"final_signal": _sig_int(report.final_signal),
|
|
"final_score": _finite(report.final_score),
|
|
"final_confidence": _finite(report.final_confidence),
|
|
"quality": str(report.signal_quality or ""),
|
|
"mtf_agreement": bool(report.mtf_agreement),
|
|
"mtf_signals": {str(k): _sig_int(v)
|
|
for k, v in (report.mtf_signals or {}).items()},
|
|
"mtf_scores": {str(k): _finite(v)
|
|
for k, v in (report.mtf_scores or {}).items()},
|
|
"penalty_tags": {str(k): _finite(v)
|
|
for k, v in (report.penalty_tags or {}).items()},
|
|
"penalty_total": _finite(report.penalty_total),
|
|
"momentum_fading": bool(report.momentum_fading),
|
|
"exhaustion_count": int(report.exhaustion_count or 0),
|
|
"spread_estimate": _finite(report.spread_estimate),
|
|
"current_price": _finite(report.current_price),
|
|
"warnings": [str(w) for w in (report.warnings or [])],
|
|
"cleaned_warnings": [str(w) for w in (report.cleaned_warnings or [])],
|
|
"bullish_reasons": [str(r) for r in (report.bullish_reasons or [])],
|
|
"bearish_reasons": [str(r) for r in (report.bearish_reasons or [])],
|
|
"setup": {
|
|
"strategy": str(getattr(setup, "strategy", "") or ""),
|
|
"setup_pattern": (str(setup.setup_pattern)
|
|
if getattr(setup, "setup_pattern", None) else None),
|
|
"round_trip_cost": _finite(getattr(setup, "round_trip_cost", None)),
|
|
"trail_arm_r": _finite(getattr(setup, "trail_arm_r", None)),
|
|
"trail_dist_r": _finite(getattr(setup, "trail_dist_r", None)),
|
|
"entry_price": _finite(getattr(setup, "entry_price", None)),
|
|
"stop_loss": _finite(getattr(setup, "stop_loss", None)),
|
|
"take_profit_1": _finite(getattr(setup, "take_profit_1", None)),
|
|
"take_profit_2": _finite(getattr(setup, "take_profit_2", None)),
|
|
"risk_reward": _finite(getattr(setup, "risk_reward", None)),
|
|
},
|
|
"tfs": [_tf_block(analysis, role, direction)
|
|
for role, analysis in _role_analyses(report, confirmation_tfs)],
|
|
}
|
|
|
|
|
|
def composite_from_rows(eff_weights: Iterable, wscores: Iterable) -> Tuple[float, int]:
|
|
"""Reproduce (composite score, signal int) from stored indicator rows."""
|
|
ew = np.asarray(list(eff_weights), dtype=float)
|
|
ws = np.asarray(list(wscores), dtype=float)
|
|
total_w = float(np.nansum(ew)) if ew.size else 0.0
|
|
total_s = float(np.nansum(ws)) if ws.size else 0.0
|
|
if total_w == 0:
|
|
return 0.0, 0
|
|
score = max(-100.0, min(100.0, (total_s / total_w) * 50.0))
|
|
strong = COMPOSITE_THRESHOLDS["strong"]
|
|
weak = COMPOSITE_THRESHOLDS["weak"]
|
|
if score >= strong:
|
|
sig = 2
|
|
elif score >= weak:
|
|
sig = 1
|
|
elif score <= -strong:
|
|
sig = -2
|
|
elif score <= -weak:
|
|
sig = -1
|
|
else:
|
|
sig = 0
|
|
return score, sig
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
# TRADE + INDICATOR TABLES
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
|
|
def _blocks_by_role(snap: Dict) -> Dict[str, Dict]:
|
|
return {str(b.get("role", "")): b for b in (snap.get("tfs") or [])}
|
|
|
|
|
|
def build_trades_table(trades, book_df: pd.DataFrame,
|
|
primary_tf: Timeframe) -> pd.DataFrame:
|
|
"""One row per accepted trade: the 63 book columns + decision context.
|
|
|
|
`book_df` is `gpu_backtest.trades_to_dataframe(trades)` (passed in so
|
|
this module never imports the backtester) and is copied verbatim, in
|
|
order, so a recorded run is a strict superset of the pinned book.
|
|
"""
|
|
trades = list(trades or [])
|
|
book = (book_df.copy().reset_index(drop=True)
|
|
if book_df is not None else pd.DataFrame())
|
|
book_cols = list(book.columns)
|
|
|
|
warned_empty = False
|
|
pen_tags: List[str] = []
|
|
seen_tags = set()
|
|
for t in trades:
|
|
for tag in (getattr(t, "snapshot", None) or {}).get("penalty_tags", {}) or {}:
|
|
if tag not in seen_tags:
|
|
seen_tags.add(tag)
|
|
pen_tags.append(str(tag))
|
|
|
|
rows: List[Dict] = []
|
|
bar_ts_raw: List[Optional[str]] = []
|
|
dec_ts_raw: List[Optional[str]] = []
|
|
|
|
for i, t in enumerate(trades):
|
|
snap = getattr(t, "snapshot", None) or {}
|
|
if not snap and not warned_empty:
|
|
warned_empty = True
|
|
logger.warning(
|
|
"--record: trade snapshots are empty — decision context "
|
|
"columns will be NaN/'' (was the run recorded?)")
|
|
blocks = _blocks_by_role(snap)
|
|
setup = snap.get("setup") or {}
|
|
bar_ts_raw.append(snap.get("bar_ts") or None)
|
|
dec_ts_raw.append(snap.get("decision_time") or None)
|
|
|
|
row: Dict = {
|
|
"tp2": _finite(getattr(t, "tp2", None)),
|
|
"risk_pips": _finite(getattr(t, "risk_pips", None)),
|
|
"reward_pips": _finite(getattr(t, "reward_pips", None)),
|
|
"pip_value": _finite(getattr(t, "pip_value", None)),
|
|
"stake_multiplier": _finite(getattr(t, "stake_multiplier", None)),
|
|
}
|
|
for role in ROLES:
|
|
b = blocks.get(role)
|
|
if b is None:
|
|
row.update({
|
|
f"{role}_tf": "", f"{role}_regime": "", f"{role}_vol": "",
|
|
f"{role}_composite": np.nan, f"{role}_composite_signal": pd.NA,
|
|
f"{role}_confidence": np.nan, f"{role}_atr": np.nan,
|
|
f"{role}_n_voting": pd.NA,
|
|
f"{role}_sum_eff_weight": np.nan,
|
|
f"{role}_sum_wscore": np.nan,
|
|
})
|
|
continue
|
|
inds = b.get("indicators") or {}
|
|
effs = [v.get("eff_weight") for v in inds.values()]
|
|
wss = [v.get("wscore") for v in inds.values()]
|
|
row.update({
|
|
f"{role}_tf": str(b.get("tf") or ""),
|
|
f"{role}_regime": str(b.get("regime") or ""),
|
|
f"{role}_vol": str(b.get("vol") or ""),
|
|
f"{role}_composite": _num(b.get("composite_score")),
|
|
f"{role}_composite_signal": int(b.get("composite_signal") or 0),
|
|
f"{role}_confidence": _num(b.get("confidence")),
|
|
f"{role}_atr": _num(b.get("atr")),
|
|
f"{role}_n_voting": len(inds),
|
|
f"{role}_sum_eff_weight": float(
|
|
np.nansum([_num(x) for x in effs])) if effs else np.nan,
|
|
f"{role}_sum_wscore": float(
|
|
np.nansum([_num(x) for x in wss])) if wss else np.nan,
|
|
})
|
|
|
|
row["mtf_agreement"] = bool(snap.get("mtf_agreement", False))
|
|
row["final_signal"] = int(snap.get("final_signal") or 0)
|
|
row["penalty_total"] = _num(snap.get("penalty_total"))
|
|
tags = snap.get("penalty_tags") or {}
|
|
for tag in pen_tags:
|
|
row[f"pen_{_sanitize(tag)}"] = _num(tags.get(tag)) if tag in tags else np.nan
|
|
row["exhaustion_count"] = int(snap.get("exhaustion_count") or 0)
|
|
row["momentum_fading"] = bool(snap.get("momentum_fading", False))
|
|
|
|
row["setup_strategy"] = str(setup.get("strategy") or "")
|
|
row["setup_pattern"] = str(setup.get("setup_pattern") or "")
|
|
row["round_trip_cost"] = _num(setup.get("round_trip_cost"))
|
|
row["setup_trail_arm_r"] = _num(setup.get("trail_arm_r"))
|
|
row["setup_trail_dist_r"] = _num(setup.get("trail_dist_r"))
|
|
|
|
warns = snap.get("warnings") or []
|
|
row["warnings_all"] = " || ".join(str(w) for w in warns)
|
|
row["cleaned_warnings"] = " || ".join(
|
|
str(w) for w in (snap.get("cleaned_warnings") or []))
|
|
row["n_warnings"] = len(warns)
|
|
row["bullish_reasons_all"] = " || ".join(
|
|
str(r) for r in (snap.get("bullish_reasons") or []))
|
|
row["bearish_reasons_all"] = " || ".join(
|
|
str(r) for r in (snap.get("bearish_reasons") or []))
|
|
row["spread_estimate"] = _num(snap.get("spread_estimate"))
|
|
row["current_price"] = _num(snap.get("current_price"))
|
|
rows.append(row)
|
|
|
|
extra_cols = _trade_extra_columns(pen_tags)
|
|
extras = pd.DataFrame(rows, columns=extra_cols) if rows else \
|
|
pd.DataFrame({c: pd.Series(dtype="object") for c in extra_cols})
|
|
|
|
head = pd.DataFrame({
|
|
"trade_id": pd.Series(np.arange(len(trades)), dtype="int32"),
|
|
"bar_ts": pd.to_datetime(pd.Series(bar_ts_raw, dtype="object"),
|
|
utc=True, errors="coerce"),
|
|
"decision_time": pd.to_datetime(pd.Series(dec_ts_raw, dtype="object"),
|
|
utc=True, errors="coerce"),
|
|
})
|
|
if not rows:
|
|
head = pd.DataFrame({
|
|
"trade_id": pd.Series(dtype="int32"),
|
|
"bar_ts": pd.Series(dtype="datetime64[ns, UTC]"),
|
|
"decision_time": pd.Series(dtype="datetime64[ns, UTC]"),
|
|
})
|
|
|
|
df = pd.concat(
|
|
[head.reset_index(drop=True), book.reset_index(drop=True),
|
|
extras.reset_index(drop=True)], axis=1)
|
|
df = df[["trade_id", "bar_ts", "decision_time"] + book_cols + extra_cols]
|
|
|
|
# Compact dtypes (nullable where a value can legitimately be absent).
|
|
if len(df):
|
|
for role in ROLES:
|
|
df[f"{role}_composite_signal"] = df[f"{role}_composite_signal"].astype("Int8")
|
|
df[f"{role}_n_voting"] = df[f"{role}_n_voting"].astype("Int16")
|
|
df["final_signal"] = df["final_signal"].astype("int8")
|
|
df["exhaustion_count"] = df["exhaustion_count"].astype("int16")
|
|
df["n_warnings"] = df["n_warnings"].astype("int8")
|
|
df["mtf_agreement"] = df["mtf_agreement"].astype(bool)
|
|
df["momentum_fading"] = df["momentum_fading"].astype(bool)
|
|
return df
|
|
|
|
|
|
def _trade_extra_columns(pen_tags: List[str]) -> List[str]:
|
|
cols = ["tp2", "risk_pips", "reward_pips", "pip_value", "stake_multiplier"]
|
|
for role in ROLES:
|
|
cols += [f"{role}_tf", f"{role}_regime", f"{role}_vol",
|
|
f"{role}_composite", f"{role}_composite_signal",
|
|
f"{role}_confidence", f"{role}_atr", f"{role}_n_voting",
|
|
f"{role}_sum_eff_weight", f"{role}_sum_wscore"]
|
|
cols += ["mtf_agreement", "final_signal", "penalty_total"]
|
|
cols += [f"pen_{_sanitize(t)}" for t in pen_tags]
|
|
cols += ["exhaustion_count", "momentum_fading", "setup_strategy",
|
|
"setup_pattern", "round_trip_cost", "setup_trail_arm_r",
|
|
"setup_trail_dist_r", "warnings_all", "cleaned_warnings",
|
|
"n_warnings", "bullish_reasons_all", "bearish_reasons_all",
|
|
"spread_estimate", "current_price"]
|
|
return cols
|
|
|
|
|
|
def build_indicators_table(trades) -> pd.DataFrame:
|
|
"""LONG table: one row per (trade, timeframe role, indicator that VOTED).
|
|
|
|
Divergence and candlestick only exist when they fired — an absent row
|
|
means "no vote", never a zero vote.
|
|
"""
|
|
rows: List[Dict] = []
|
|
for i, t in enumerate(list(trades or [])):
|
|
snap = getattr(t, "snapshot", None) or {}
|
|
for block in (snap.get("tfs") or []):
|
|
regime = str(block.get("regime") or "")
|
|
tf_name = str(block.get("tf") or "")
|
|
role = str(block.get("role") or "")
|
|
for name, ind in (block.get("indicators") or {}).items():
|
|
rows.append({
|
|
"trade_id": i,
|
|
"tf_role": role,
|
|
"timeframe": tf_name,
|
|
"indicator": str(name),
|
|
"signal": int(ind.get("signal") or 0),
|
|
"signal_name": str(ind.get("signal_name") or ""),
|
|
"value": _num(ind.get("value")),
|
|
"confidence": _num(ind.get("confidence")),
|
|
"base_weight": _num(ind.get("base_weight")),
|
|
"regime_mult": _num(ind.get("regime_mult")),
|
|
"eff_weight": _num(ind.get("eff_weight")),
|
|
"wscore": _num(ind.get("wscore")),
|
|
"agrees": str(ind.get("agrees") or ""),
|
|
"regime": regime,
|
|
"description": str(ind.get("description") or ""),
|
|
})
|
|
if not rows:
|
|
return pd.DataFrame({c: pd.Series(dtype="object") for c in _IND_COLS})
|
|
df = pd.DataFrame(rows, columns=_IND_COLS)
|
|
df["trade_id"] = df["trade_id"].astype("int32")
|
|
df["signal"] = df["signal"].astype("int8")
|
|
for col in ("value", "confidence", "base_weight", "regime_mult",
|
|
"eff_weight", "wscore"):
|
|
df[col] = df[col].astype("float64")
|
|
return df
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
# ECONOMIC CALENDAR
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
|
|
def load_calendar(path=CALENDAR_PARQUET) -> pd.DataFrame:
|
|
"""Forex-Factory calendar parquet, normalised.
|
|
|
|
`timeLabel` is a display string and is NEVER used for timing —
|
|
`dateline` (unix seconds UTC) is the only clock. `timeMasked` marks
|
|
all-day rows, whose `dateline` is only a date anchor.
|
|
"""
|
|
cal = pd.read_parquet(path)
|
|
cal = cal.copy()
|
|
cal["ts_utc"] = pd.to_datetime(
|
|
cal["dateline"], unit="s", utc=True).astype("datetime64[ns, UTC]")
|
|
cal["timed"] = ~cal["timeMasked"].astype(bool)
|
|
cal["et_date"] = cal["ts_utc"].dt.tz_convert(_NY).dt.date
|
|
cal = cal.rename(columns={"impactName": "impact",
|
|
"actualBetterWorse": "better_worse"})
|
|
for col in ("actual", "forecast", "previous", "revision"):
|
|
if col in cal.columns:
|
|
cal[col] = cal[col].fillna("").astype(str)
|
|
return cal.sort_values("ts_utc").reset_index(drop=True)
|
|
|
|
|
|
def calendar_fingerprint(path=CALENDAR_PARQUET) -> Optional[Dict]:
|
|
p = Path(path)
|
|
if not p.exists():
|
|
return None
|
|
try:
|
|
cal = load_calendar(p)
|
|
return {
|
|
"path": str(p),
|
|
"md5": hashlib.md5(p.read_bytes()).hexdigest(),
|
|
"n_rows": int(len(cal)),
|
|
"min_ts": cal["ts_utc"].min().isoformat() if len(cal) else None,
|
|
"max_ts": cal["ts_utc"].max().isoformat() if len(cal) else None,
|
|
}
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning("--record: calendar fingerprint failed: %s", exc)
|
|
return None
|
|
|
|
|
|
def _ns(series: pd.Series) -> np.ndarray:
|
|
"""tz-aware datetimes -> int64 nanoseconds (NaT -> iinfo.min)."""
|
|
s = pd.to_datetime(series, utc=True, errors="coerce").astype("datetime64[ns, UTC]")
|
|
return s.astype("int64").to_numpy()
|
|
|
|
|
|
_NAT_NS = np.iinfo(np.int64).min
|
|
|
|
|
|
def _empty(cols: List[str]) -> pd.DataFrame:
|
|
return pd.DataFrame({c: pd.Series(dtype="object") for c in cols})
|
|
|
|
|
|
def build_events_table(trades_df: pd.DataFrame, cal: pd.DataFrame) -> pd.DataFrame:
|
|
"""LONG table: calendar rows around each trade.
|
|
|
|
Timed rows land in the window ``[entry - 4h, max(exit, entry + 4h)]``
|
|
(both ends inclusive) and are phased `before` / `during_hold` /
|
|
`after_exit`. All-day (masked) rows are attached when their New-York
|
|
calendar date equals the entry's New-York date.
|
|
"""
|
|
if trades_df is None or trades_df.empty or cal is None or cal.empty:
|
|
return _empty(_EVENT_COLS)
|
|
|
|
timed = cal[cal["timed"]].reset_index(drop=True)
|
|
masked = cal[~cal["timed"]].reset_index(drop=True)
|
|
t_ns = _ns(timed["ts_utc"]) if len(timed) else np.zeros(0, dtype=np.int64)
|
|
|
|
by_date: Dict[object, List[int]] = {}
|
|
for i, d in enumerate(masked["et_date"].tolist()):
|
|
by_date.setdefault(d, []).append(i)
|
|
|
|
entry_ns = _ns(trades_df["entry_time"])
|
|
exit_ns = (_ns(trades_df["exit_time"]) if "exit_time" in trades_df.columns
|
|
else np.full(len(trades_df), _NAT_NS, dtype=np.int64))
|
|
entry_et = pd.to_datetime(trades_df["entry_time"], utc=True) \
|
|
.dt.tz_convert(_NY).dt.date.tolist()
|
|
trade_ids = (trades_df["trade_id"].tolist() if "trade_id" in trades_df.columns
|
|
else list(range(len(trades_df))))
|
|
win_ns = int(EVENT_WINDOW.value)
|
|
|
|
rows: List[Dict] = []
|
|
for i in range(len(trades_df)):
|
|
e = int(entry_ns[i])
|
|
if e == _NAT_NS:
|
|
continue
|
|
x = int(exit_ns[i])
|
|
has_exit = x != _NAT_NS
|
|
# No exit yet -> treat the hold as zero-length; the window still
|
|
# runs to entry + 4h so the release context is recorded.
|
|
x_eff = x if has_exit else e
|
|
lo = e - win_ns
|
|
hi = max(x_eff, e + win_ns)
|
|
if len(t_ns):
|
|
a = int(np.searchsorted(t_ns, lo, side="left"))
|
|
b = int(np.searchsorted(t_ns, hi, side="right"))
|
|
else:
|
|
a = b = 0
|
|
for j in range(a, b):
|
|
ts = int(t_ns[j])
|
|
if ts < e:
|
|
phase = "before"
|
|
elif ts <= x_eff:
|
|
phase = "during_hold"
|
|
else:
|
|
phase = "after_exit"
|
|
r = timed.iloc[j]
|
|
rows.append({
|
|
"trade_id": trade_ids[i],
|
|
"event_id": int(r["id"]),
|
|
"ts_utc": r["ts_utc"],
|
|
"name": str(r["name"]),
|
|
"currency": str(r["currency"]),
|
|
"impact": str(r["impact"]),
|
|
"timed": True,
|
|
"actual": str(r["actual"]),
|
|
"forecast": str(r["forecast"]),
|
|
"previous": str(r["previous"]),
|
|
"revision": str(r["revision"]),
|
|
"better_worse": int(r["better_worse"]),
|
|
"offset_min": (ts - e) / 6e10,
|
|
"phase": phase,
|
|
})
|
|
for j in by_date.get(entry_et[i], ()):
|
|
r = masked.iloc[j]
|
|
rows.append({
|
|
"trade_id": trade_ids[i],
|
|
"event_id": int(r["id"]),
|
|
"ts_utc": r["ts_utc"],
|
|
"name": str(r["name"]),
|
|
"currency": str(r["currency"]),
|
|
"impact": str(r["impact"]),
|
|
"timed": False,
|
|
"actual": str(r["actual"]),
|
|
"forecast": str(r["forecast"]),
|
|
"previous": str(r["previous"]),
|
|
"revision": str(r["revision"]),
|
|
"better_worse": int(r["better_worse"]),
|
|
"offset_min": np.nan,
|
|
"phase": "all_day",
|
|
})
|
|
|
|
if not rows:
|
|
return _empty(_EVENT_COLS)
|
|
df = pd.DataFrame(rows, columns=_EVENT_COLS)
|
|
df["event_id"] = df["event_id"].astype("int64")
|
|
df["better_worse"] = df["better_worse"].astype("int8")
|
|
df["offset_min"] = df["offset_min"].astype("float64")
|
|
df["timed"] = df["timed"].astype(bool)
|
|
return df
|
|
|
|
|
|
def calendar_scalars(trades_df: pd.DataFrame, cal: pd.DataFrame) -> pd.DataFrame:
|
|
"""Per-trade calendar summary, index-aligned to `trades_df`."""
|
|
n = 0 if trades_df is None else len(trades_df)
|
|
out = pd.DataFrame({
|
|
"ev_prev_hi_usd_min": np.full(n, np.nan),
|
|
"ev_prev_hi_usd_name": [""] * n,
|
|
"ev_next_hi_usd_min": np.full(n, np.nan),
|
|
"ev_next_hi_usd_name": [""] * n,
|
|
"ev_hi_usd_during_hold": np.zeros(n, dtype="int16"),
|
|
"ev_hi_usd_during_hold_names": [""] * n,
|
|
"ev_any_hi_during_hold": np.zeros(n, dtype="int16"),
|
|
"ev_entry_within_30m_hi_usd": np.zeros(n, dtype=bool),
|
|
"us_holiday": np.zeros(n, dtype=bool),
|
|
}, index=(trades_df.index if n else None))
|
|
if not n or cal is None or cal.empty:
|
|
return out
|
|
|
|
hi_all = cal[(cal["impact"] == "high") & cal["timed"]].reset_index(drop=True)
|
|
hi_usd = hi_all[hi_all["currency"] == "USD"].reset_index(drop=True)
|
|
hi_all_ns = _ns(hi_all["ts_utc"]) if len(hi_all) else np.zeros(0, np.int64)
|
|
hi_usd_ns = _ns(hi_usd["ts_utc"]) if len(hi_usd) else np.zeros(0, np.int64)
|
|
usd_names = hi_usd["name"].astype(str).tolist()
|
|
|
|
hol = cal[(cal["currency"] == "USD") & (cal["impact"] == "holiday")
|
|
& (~cal["timed"])]
|
|
hol_dates = set(hol["et_date"].tolist())
|
|
|
|
entry_ns = _ns(trades_df["entry_time"])
|
|
exit_ns = (_ns(trades_df["exit_time"]) if "exit_time" in trades_df.columns
|
|
else np.full(n, _NAT_NS, dtype=np.int64))
|
|
entry_et = pd.to_datetime(trades_df["entry_time"], utc=True) \
|
|
.dt.tz_convert(_NY).dt.date.tolist()
|
|
half_hour = int(30 * 6e10)
|
|
|
|
for i in range(n):
|
|
e = int(entry_ns[i])
|
|
if e == _NAT_NS:
|
|
continue
|
|
pos = int(np.searchsorted(hi_usd_ns, e, side="right"))
|
|
if pos > 0:
|
|
mins = (e - int(hi_usd_ns[pos - 1])) / 6e10
|
|
if mins <= NEAR_RELEASE_MAX_MIN:
|
|
out.iat[i, 0] = mins
|
|
out.iat[i, 1] = usd_names[pos - 1]
|
|
if pos < len(hi_usd_ns):
|
|
mins = (int(hi_usd_ns[pos]) - e) / 6e10
|
|
if mins <= NEAR_RELEASE_MAX_MIN:
|
|
out.iat[i, 2] = mins
|
|
out.iat[i, 3] = usd_names[pos]
|
|
|
|
x = int(exit_ns[i])
|
|
if x != _NAT_NS and x > e:
|
|
lo = int(np.searchsorted(hi_usd_ns, e, side="right"))
|
|
hi = int(np.searchsorted(hi_usd_ns, x, side="right"))
|
|
out.iat[i, 4] = int(max(0, hi - lo))
|
|
out.iat[i, 5] = "|".join(usd_names[lo:hi])
|
|
lo_a = int(np.searchsorted(hi_all_ns, e, side="right"))
|
|
hi_a = int(np.searchsorted(hi_all_ns, x, side="right"))
|
|
out.iat[i, 6] = int(max(0, hi_a - lo_a))
|
|
|
|
if len(hi_usd_ns):
|
|
a = int(np.searchsorted(hi_usd_ns, e - half_hour, side="left"))
|
|
b = int(np.searchsorted(hi_usd_ns, e + half_hour, side="right"))
|
|
out.iat[i, 7] = bool(b > a)
|
|
out.iat[i, 8] = entry_et[i] in hol_dates
|
|
|
|
out["ev_hi_usd_during_hold"] = out["ev_hi_usd_during_hold"].astype("int16")
|
|
out["ev_any_hi_during_hold"] = out["ev_any_hi_during_hold"].astype("int16")
|
|
out["ev_entry_within_30m_hi_usd"] = out["ev_entry_within_30m_hi_usd"].astype(bool)
|
|
out["us_holiday"] = out["us_holiday"].astype(bool)
|
|
return out
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
# PER-BAR RECORDER (--record-bars)
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
|
|
def new_bar_buffer() -> Dict:
|
|
return {"rows": [], "pos": [], "ts_ns": [], "warning0": []}
|
|
|
|
|
|
def add_bar_row(buf: Dict, pos: int, ts, report,
|
|
primary_tf, confirmation_tfs) -> None:
|
|
"""Append one compact float32 row for an ANALYZED primary bar.
|
|
|
|
Runs ~140k times on a 2-year M5 replay, so it allocates exactly one
|
|
NaN row and fills it through a precomputed column-index map. An
|
|
absent indicator or a missing confirmation block stays NaN — never 0
|
|
(0 is a real NEUTRAL vote).
|
|
"""
|
|
row = np.full(_N_BAR_COLS, np.nan, dtype=np.float32)
|
|
idx = _BAR_IDX
|
|
setup = getattr(report, "trade_setup", None)
|
|
|
|
row[idx["close"]] = _num(report.current_price)
|
|
row[idx["final_score"]] = _num(report.final_score)
|
|
row[idx["final_conf"]] = _num(report.final_confidence)
|
|
row[idx["quality_code"]] = QUALITY_CODES.get(
|
|
str(report.signal_quality or ""), np.nan)
|
|
row[idx["strategy_code"]] = (
|
|
STRATEGY_CODES.get(str(getattr(setup, "strategy", "") or ""), np.nan)
|
|
if setup is not None else 0)
|
|
row[idx["has_setup"]] = 1.0 if setup is not None else 0.0
|
|
row[idx["final_signal"]] = _sig_int(report.final_signal)
|
|
row[idx["mtf_agreement"]] = 1.0 if report.mtf_agreement else 0.0
|
|
row[idx["penalty_total"]] = _num(report.penalty_total)
|
|
row[idx["exhaustion_count"]] = int(report.exhaustion_count or 0)
|
|
|
|
for role, analysis in _role_analyses(report, confirmation_tfs):
|
|
levels = getattr(analysis, "key_levels", None) or {}
|
|
row[idx[f"{role}_regime_code"]] = REGIME_CODES.get(
|
|
str(getattr(analysis, "market_regime", "") or ""), np.nan)
|
|
row[idx[f"{role}_vol_code"]] = VOL_CODES.get(
|
|
str(getattr(analysis, "volatility_regime", "") or ""), np.nan)
|
|
row[idx[f"{role}_composite"]] = _num(getattr(analysis, "composite_score", None))
|
|
row[idx[f"{role}_confidence"]] = _num(getattr(analysis, "confidence", None))
|
|
row[idx[f"{role}_atr"]] = _num(levels.get("atr"))
|
|
for name, ind in (getattr(analysis, "indicators", None) or {}).items():
|
|
if name.startswith("_") or ind.weight == 0:
|
|
continue
|
|
key = f"{role}_{name}_"
|
|
base = idx.get(key + "signal")
|
|
if base is None:
|
|
continue # indicator outside INDICATORS
|
|
row[base] = _sig_int(ind.signal)
|
|
row[base + 1] = _num(ind.value)
|
|
row[base + 2] = _num(ind.confidence)
|
|
row[base + 3] = _num(getattr(ind, "effective_weight", None))
|
|
row[base + 4] = _num(getattr(ind, "weighted_score", None))
|
|
|
|
buf["rows"].append(row)
|
|
buf["pos"].append(int(pos))
|
|
buf["ts_ns"].append(int(pd.Timestamp(ts).value))
|
|
warns = report.warnings or []
|
|
buf["warning0"].append(str(warns[0]) if warns else "")
|
|
|
|
|
|
def finish_bar_rows(buf: Optional[Dict]):
|
|
"""(matrix float32 [n,K], ts_ns int64, pos int64, warning0 list) or None."""
|
|
if not buf or not buf.get("rows"):
|
|
return None
|
|
mat = np.vstack(buf["rows"]).astype(np.float32, copy=False)
|
|
return (mat,
|
|
np.asarray(buf["ts_ns"], dtype=np.int64),
|
|
np.asarray(buf["pos"], dtype=np.int64),
|
|
list(buf["warning0"]))
|
|
|
|
|
|
def forward_moves(close: np.ndarray, high: np.ndarray, low: np.ndarray,
|
|
atr: np.ndarray, k: int) -> Dict[str, np.ndarray]:
|
|
"""ATR-normalised forward move / MFE / MAE over the next `k` bars.
|
|
|
|
NaN where the horizon runs off the end of the series or the anchor
|
|
bar has no usable ATR.
|
|
"""
|
|
close = np.asarray(close, dtype=np.float64)
|
|
high = np.asarray(high, dtype=np.float64)
|
|
low = np.asarray(low, dtype=np.float64)
|
|
atr = np.asarray(atr, dtype=np.float64)
|
|
n = close.size
|
|
fwd = np.full(n, np.nan)
|
|
mfe = np.full(n, np.nan)
|
|
mae = np.full(n, np.nan)
|
|
if k <= 0 or n <= k:
|
|
return {"fwd": fwd, "mfe": mfe, "mae": mae}
|
|
|
|
m = n - k # anchors 0 .. n-k-1
|
|
with np.errstate(invalid="ignore", divide="ignore"):
|
|
denom = atr[:m]
|
|
fwd[:m] = (close[k:] - close[:m]) / denom
|
|
win_hi = np.lib.stride_tricks.sliding_window_view(high[1:], k)[:m]
|
|
win_lo = np.lib.stride_tricks.sliding_window_view(low[1:], k)[:m]
|
|
mfe[:m] = (win_hi.max(axis=1) - close[:m]) / denom
|
|
mae[:m] = (close[:m] - win_lo.min(axis=1)) / denom
|
|
|
|
bad = ~np.isfinite(atr) | (atr <= 0)
|
|
fwd[bad] = np.nan
|
|
mfe[bad] = np.nan
|
|
mae[bad] = np.nan
|
|
return {"fwd": fwd, "mfe": mfe, "mae": mae}
|
|
|
|
|
|
def build_bars_table(chunks, sim_df: pd.DataFrame, primary_tf: Timeframe,
|
|
primary_atr: Optional[np.ndarray] = None) -> Optional[pd.DataFrame]:
|
|
"""Concat worker bar chunks into one frame + ATR-normalised forward moves."""
|
|
mats, tss, poss = [], [], []
|
|
warns: List[str] = []
|
|
for chunk in (chunks or []):
|
|
if chunk is None:
|
|
continue
|
|
mat, ts_ns, pos, warn0 = chunk
|
|
mats.append(mat)
|
|
tss.append(ts_ns)
|
|
poss.append(pos)
|
|
warns.extend(warn0)
|
|
if not mats:
|
|
return None
|
|
|
|
mat = np.vstack(mats)
|
|
ts_ns = np.concatenate(tss)
|
|
pos = np.concatenate(poss)
|
|
order = np.argsort(pos, kind="stable")
|
|
mat = mat[order]
|
|
ts_ns = ts_ns[order]
|
|
pos = pos[order]
|
|
warns = [warns[i] for i in order]
|
|
|
|
bar_ts = pd.to_datetime(ts_ns, utc=True)
|
|
df = pd.DataFrame(mat, columns=list(BAR_COLS))
|
|
df.insert(0, "warning0", warns)
|
|
df.insert(0, "raw_pos", np.zeros(len(df), dtype=np.int32))
|
|
df.insert(0, "pos", pos.astype(np.int32))
|
|
df.insert(0, "bar_ts", bar_ts)
|
|
|
|
if sim_df is None or sim_df.empty:
|
|
return df
|
|
|
|
raw_pos = np.asarray(sim_df.index.get_indexer(bar_ts), dtype=np.int64)
|
|
df["raw_pos"] = raw_pos.astype(np.int32)
|
|
valid = raw_pos >= 0
|
|
|
|
close = sim_df["Close"].to_numpy(dtype=np.float64)
|
|
high = sim_df["High"].to_numpy(dtype=np.float64)
|
|
low = sim_df["Low"].to_numpy(dtype=np.float64)
|
|
n_raw = close.size
|
|
|
|
atr_vals = (np.asarray(primary_atr, dtype=np.float64)
|
|
if primary_atr is not None
|
|
else df["primary_atr"].to_numpy(dtype=np.float64))
|
|
atr_full = np.full(n_raw, np.nan)
|
|
if valid.any():
|
|
atr_full[raw_pos[valid]] = atr_vals[valid]
|
|
|
|
bar_seconds = int(pd.Timedelta(primary_tf.value).total_seconds())
|
|
horizons = {"1h": 3600, "4h": 4 * 3600, "24h": 24 * 3600}
|
|
ks = {label: max(1, secs // max(bar_seconds, 1))
|
|
for label, secs in horizons.items()}
|
|
|
|
def _pick(vec: np.ndarray) -> np.ndarray:
|
|
out = np.full(len(df), np.nan)
|
|
if valid.any():
|
|
out[valid] = vec[raw_pos[valid]]
|
|
return out
|
|
|
|
moves = {label: forward_moves(close, high, low, atr_full, k)
|
|
for label, k in ks.items()}
|
|
df["fwd_1h_atr"] = _pick(moves["1h"]["fwd"])
|
|
df["fwd_4h_atr"] = _pick(moves["4h"]["fwd"])
|
|
df["fwd_24h_atr"] = _pick(moves["24h"]["fwd"])
|
|
df["mfe_4h_atr"] = _pick(moves["4h"]["mfe"])
|
|
df["mae_4h_atr"] = _pick(moves["4h"]["mae"])
|
|
df["mfe_24h_atr"] = _pick(moves["24h"]["mfe"])
|
|
df["mae_24h_atr"] = _pick(moves["24h"]["mae"])
|
|
|
|
k24 = ks["24h"]
|
|
avail = np.full(len(df), -1, dtype=np.int64)
|
|
if valid.any():
|
|
avail[valid] = np.minimum(k24, n_raw - 1 - raw_pos[valid])
|
|
df["fwd_bars_24h"] = avail.astype("int32")
|
|
return df
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
# RUN WRITER / READER
|
|
# ═════════════════════════════════════════════════════════════════════
|
|
|
|
def _normalise_datetimes(df: pd.DataFrame) -> pd.DataFrame:
|
|
for col in df.columns:
|
|
if col in _DT_COLS or pd.api.types.is_datetime64_any_dtype(df[col]):
|
|
df[col] = pd.to_datetime(
|
|
df[col], utc=True, errors="coerce").astype("datetime64[ns, UTC]")
|
|
return df
|
|
|
|
|
|
def _write(df: Optional[pd.DataFrame], path: Path) -> Optional[int]:
|
|
if df is None:
|
|
return None
|
|
df = _normalise_datetimes(df.copy())
|
|
df.to_parquet(path, index=False)
|
|
return int(len(df))
|
|
|
|
|
|
def write_run(record_dir, *, trades, book_df, bt_cfg, platform_cfg, aggregator,
|
|
run_info: Dict, sim_df=None, bar_chunks=None) -> Path:
|
|
"""Write manifest.json + the parquet tables for one replay."""
|
|
out = Path(record_dir)
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
run_info = dict(run_info or {})
|
|
|
|
trades_df = build_trades_table(trades, book_df, bt_cfg.primary_tf)
|
|
ind_df = build_indicators_table(trades)
|
|
|
|
cal_fp = calendar_fingerprint(CALENDAR_PARQUET)
|
|
events_df: Optional[pd.DataFrame] = None
|
|
if cal_fp is not None:
|
|
try:
|
|
cal = load_calendar(CALENDAR_PARQUET)
|
|
events_df = build_events_table(trades_df, cal)
|
|
scalars = calendar_scalars(trades_df, cal)
|
|
for col in _SCALAR_COLS:
|
|
trades_df[col] = scalars[col].to_numpy() if len(scalars) else scalars[col]
|
|
except Exception as exc:
|
|
logger.warning("--record: calendar join failed (%s) — "
|
|
"events.parquet skipped", exc)
|
|
events_df = None
|
|
cal_fp = None
|
|
else:
|
|
logger.warning("--record: %s not found — no events table",
|
|
CALENDAR_PARQUET)
|
|
|
|
engine_fp = aggregator.engine_config_fingerprint()
|
|
gate_fp = aggregator.gate_config_fingerprint()
|
|
trades_df["engine_fingerprint"] = engine_fp
|
|
trades_df["gate_fingerprint"] = gate_fp
|
|
|
|
bars_df = (build_bars_table(bar_chunks, sim_df, bt_cfg.primary_tf)
|
|
if bar_chunks else None)
|
|
|
|
n_trades = _write(trades_df, out / "trades.parquet")
|
|
n_ind = _write(ind_df, out / "indicators.parquet")
|
|
n_events = _write(events_df, out / "events.parquet")
|
|
n_bars = _write(bars_df, out / "bars.parquet")
|
|
|
|
base_weights: Dict[str, List[float]] = {}
|
|
if len(ind_df):
|
|
for name, grp in ind_df.groupby("indicator")["base_weight"]:
|
|
base_weights[str(name)] = sorted(
|
|
{float(v) for v in grp.dropna().unique()})
|
|
|
|
skipped = dict(run_info.get("skipped") or {})
|
|
n_in_window = run_info.get("n_bars_in_window")
|
|
n_analyzed = (int(n_in_window) - int(sum(skipped.values()))
|
|
if n_in_window is not None else None)
|
|
|
|
manifest = {
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"cli": sys.argv[1:],
|
|
"git_sha": git_sha(),
|
|
"backend": run_info.get("backend"),
|
|
"pair": bt_cfg.pair,
|
|
"display_name": platform_cfg.display_name,
|
|
"primary_tf": bt_cfg.primary_tf.value,
|
|
"confirmation_tfs": [tf.value for tf in bt_cfg.confirmation_tfs],
|
|
"window": {"start": bt_cfg.start.isoformat(), "end": bt_cfg.end.isoformat()},
|
|
"spread": bt_cfg.spread,
|
|
"quality_filter": list(bt_cfg.quality_filter),
|
|
"strategies_only": list(bt_cfg.strategies_only),
|
|
"warmup_bars": bt_cfg.warmup_bars,
|
|
"cooldown_bars": bt_cfg.cooldown_bars,
|
|
"max_hold_bars": bt_cfg.max_hold_bars,
|
|
"fixed_rr_tp": bt_cfg.fixed_rr_tp,
|
|
"workers": run_info.get("workers"),
|
|
"engine_fingerprint": engine_fp,
|
|
"gate_fingerprint": gate_fp,
|
|
"technical_config": dataclasses.asdict(platform_cfg.technical),
|
|
"signals_config": dataclasses.asdict(platform_cfg.signals),
|
|
"regime_weight_table": regime_weight_table(platform_cfg.technical),
|
|
"base_weights_observed": base_weights,
|
|
"composite_thresholds": dict(COMPOSITE_THRESHOLDS),
|
|
"bars": {
|
|
"first_pos": run_info.get("first_pos"),
|
|
"last_pos": run_info.get("last_pos"),
|
|
"n_primary_bars": run_info.get("n_primary_bars"),
|
|
"n_in_window": n_in_window,
|
|
"n_analyzed": n_analyzed,
|
|
"skipped": skipped,
|
|
},
|
|
"n_candidates": run_info.get("n_candidates"),
|
|
"n_trades": int(len(trades_df)),
|
|
"cache_files": run_info.get("cache_files"),
|
|
"calendar": cal_fp,
|
|
"tables": {"trades": n_trades, "indicators": n_ind,
|
|
"events": n_events, "bars": n_bars},
|
|
"book_csv_md5": hashlib.md5(
|
|
book_df.to_csv(index=False).encode()).hexdigest()
|
|
if book_df is not None else None,
|
|
"replay_seconds": run_info.get("replay_seconds"),
|
|
}
|
|
with open(out / "manifest.json", "w", encoding="utf-8") as fh:
|
|
json.dump(manifest, fh, indent=2, default=str)
|
|
return out
|
|
|
|
|
|
def load_run(record_dir) -> Dict:
|
|
"""Read a recorded run back: manifest + tables (None when absent)."""
|
|
path = Path(record_dir)
|
|
with open(path / "manifest.json", encoding="utf-8") as fh:
|
|
manifest = json.load(fh)
|
|
def _read(name):
|
|
p = path / name
|
|
return pd.read_parquet(p) if p.exists() else None
|
|
return {
|
|
"manifest": manifest,
|
|
"trades": _read("trades.parquet"),
|
|
"indicators": _read("indicators.parquet"),
|
|
"events": _read("events.parquet"),
|
|
"bars": _read("bars.parquet"),
|
|
}
|