Harwinder Sekhon
7936c2fe6e
Signal path
- A shipped setup always names a direction: grade C needs confidence
>= 45 (gate 6c grants B at >= 45) while BUY/SELL needs >= 50, so C/B
reports reached the alert gate as NEUTRAL with a full setup. The
backtester traded them (the replay never reads final_signal) while
Telegram rejected NEUTRAL first — 5/91 trades, +3.7R of +22.2R, in
the H1 12-month book. _attach_core_setup stamps the setup's direction
and repairs the pre-gate action text; no-setup reports stay NEUTRAL.
- analyze_pair fails the scan when ANY required frame is missing
("DATA UNAVAILABLE:", scan_kind='failed'); a lost confirmation frame
used to run with an unearned -15 MTF penalty.
Outcomes and labels
- advisory_outcome.resolve_locked_floor: a gap the database cannot fill
resolves at the locked trailing stop (GAP_LOCKED, result.floor) when
the stop had crossed the entry; the rebuild fires on both gap reasons
(rows 915/920: +1.14R / +0.84R floors; resolved sum -2.46R -> -0.48R).
- label_scan_outcomes: SCAN_LABEL_MAX_AGE_H = 72 h, stale rows never
anchor the shared bar stream, labeled_at bumped on re-examination,
_stream_since warns at its row cap (the Labor-Day pile would have
blinded scan_outcomes around 2026-09-14).
Providers and scheduling
- The daily settlement pause (Mon-Thu 17:00 New York) is a scheduled
closure for _has_latest_closed_bar (whole-span label walk-back) and
_is_data_fresh (closed seconds are not age); bot and dashboard skip
scans whose just-closed label sits in the pause (61 false stale rows
in three nights).
- fetch_range retries a failed page in place (three failures on one
window raise) and logs an ERROR for interior gaps of >3 unscheduled
bars.
- pd.Timestamp.utcnow() replaced in providers/base.py and engine.py;
requirements pin pandas<4, numpy<3.
- telegram_bot: boundary loops share one scan per bar with the breach
handler (_remember_scan keys on the analysed bar; breach defers 20 s
after an M15 boundary); the rotating file handler attaches in main()
only (the test suite used to write into logs/telegram.log).
- dashboard_publisher: catch_up_timeframes re-scans boundaries missed
while a cycle overran or the process was down.
Backtester and engine guards
- _replay_span returns the per-span GPU cache-miss count; run_backtest
sums, logs it (must be 0) and writes cache_fallbacks to the manifest.
- regime_weight_table raises on a partial regime_weights_override.
Verification: 519 tests; M5/M15 2-yr, M15 OOS, H1/H4 3-yr books
byte-identical before and after; live signals.db untouched (checked on
a copy). CLAUDE.md carries the dated entry and the blocked-hours union
semantics that every book was validated under.
Claude-Session: https://claude.ai/code/session_0139xz1J6YccndA2be6yFnEo
1107 lines
47 KiB
Python
1107 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"),
|
|
# >0 means some bar-analyses fell back to the live engine's pandas
|
|
# math instead of the pre-computed cache (a parity hazard).
|
|
"cache_fallbacks": run_info.get("cache_fallbacks"),
|
|
"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"),
|
|
}
|