harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

backtest cleanup: drop four research/one-time scripts; books and stale cache moved out (owner decisions 2026-09-15)

Browse Source 
Discussion-first cleanup of backtest/ and scripts/. Nothing in the signal
core or the backtester changed.

Removed (last commit containing them: 931b265):
- backtest/diagnose_filters.py + its two Module 18 tests
- scripts/shadow_scan.py (Phase 4 shadow scan, never run)
- scripts/backfill_news_archive.py, scripts/migrate_signal_log_news.py
  (one-time scripts; the interning test keeps its live-path half, the
  prune test goes)

Kept: the replay-recording stack (replay_record.py, replay_scorecard.py,
ff_calendar_scraper.py, cache/calendar.parquet, Module 33).

Local gitignored data was moved to ~/Development/ForexBot2_trash_2026-09-15
(346 MB) for the owner to delete: all 35 pinned backtest/*.csv books, the
three research output folders, the BTC/ETH cache snapshots, the Tiingo
parquets and 13 uncited Oanda windows. The three research-folder ignore
rules are dropped with the folders.

Verification: all five gate books regenerate offline from the pruned cache
with the recorded MD5s (M5 W2 29120a1e, M15 W2 d185f099, M15 OOS 6f8f24b4,
H1 W3 3b45d12b, H4 W3 e75adae1; NumPy CPU, 0 cache fallbacks). Suite
506 passed, 0 failed, 33 modules.
This commit is contained in:
Harwinder Sekhon
2026-09-15 20:15:06 -04:00
parent c916ecf213
commit 9adaee65d2
7 changed files with 9 additions and 1327 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
-3
View File
@@ -29,6 +29,3 @@ backtest/*.csv
backtest/__pycache__/
.env.*
logs/
backtest/research/sizing_2026-09/
backtest/research/replay_2026-09/
backtest/research/cycle_2026-09-10/
CLAUDE.md
+7 -6
View File
File diff suppressed because one or more lines are too long
backtest/diagnose_filters.py
-627
View File
@@ -1,627 +0,0 @@
"""
ForexBot2 — Filter Rejection Diagnostic
========================================
Replays the same backtest loop as gpu_backtest.py but instead of
collecting trades, counts how many M15 bars are rejected by each
gate in the signal pipeline.
Produces a summary table showing where potential trades are being
lost, helping tune filter thresholds.
Usage:
python backtest/diagnose_filters.py --pair XAUUSD --days 365
python backtest/diagnose_filters.py --pair XAUUSD --days 365 --sample-every 1
"""
from __future__ import annotations
import argparse
import logging
import sys
import time
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
# Ensure project root 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 ( # noqa: E402
PAIR_INFO,
PlatformConfig,
SignalStrength,
Timeframe,
)
from signal_aggregator import SignalAggregator, SignalReport # noqa: E402
# Reuse GPU backtest infrastructure (data loading, indicator pre-compute, etc.)
from gpu_backtest import ( # noqa: E402
BAR_SECONDS,
GPUTechnicalEngine,
PAIR_ALIASES,
IndicatorCache,
fetch_history,
precompute_indicators,
_is_xau_blocked_by_calendar,
)
from providers import make_data_provider # noqa: E402
logger = logging.getLogger("diagnose_filters")
# ═════════════════════════════════════════════════════════════════════
# OUTCOME CLASSIFICATION
# ═════════════════════════════════════════════════════════════════════
def classify_bar(
report: SignalReport,
is_xau: bool,
) -> Tuple[str, List[str]]:
"""Classify a bar's outcome and list which filters killed it.
Returns:
(outcome, [filter_names])
Outcomes:
"TRADE" — quality C+ and trade_setup exists
"QUALITY_OK_NO_SETUP" — quality C+ but trade_setup is None
"LOW_QUALITY" — signal reached BUY/SELL but quality D or F
"NEUTRAL" — signal stayed NEUTRAL (score didn't reach +/-30)
Filter names are extracted from report.warnings.
"""
filters_hit: List[str] = []
# Known filter tags, matched against the exact warning strings the
# aggregator emits (refreshed 2026-07-30 for gates 6d-6i). ORDER
# MATTERS: most-specific prefix first, and each warning maps to at
# most ONE tag (break on first hit) — substring families like
# "M5 FILTER: SELL quality" vs "M5 FILTER" and "TREND EXHAUSTION"
# vs "EXHAUSTION" would otherwise double-count.
filter_tags = [
# M5 noise gates (6i, _apply_m5_filters)
("M5 FILTER: |score|", "M5_SCORE"),
("M5 FILTER: no MTF agreement", "M5_MTF"),
("M5 FILTER: no confirmation TF", "M5_CONF"),
("M5 FILTER: SELL quality", "M5_SELL_QUALITY"),
("M5 FILTER: SELL blocked — RSI", "M5_SELL_RSI"),
("M5 FILTER: '", "M5_VOL"),
("M5 FILTER", "M5_OTHER"),
# XAU SELL gates (6e/6g/6g2/6g3)
("SELL FILTER: quality", "SELL_QUALITY"),
("SELL FILTER: volatility", "SELL_VOL"),
("SELL FILTER: regime", "SELL_REGIME"),
("UTC entry blocked for shorts", "SELL_HOUR"),
# Hard gates
("MTF OPPOSITION", "MTF_OPPOSITION"),
("EXHAUSTED TREND FILTER", "EXHAUSTED_TREND"),
("BIG CANDLE FILTER", "BIG_CANDLE"),
("WEEKEND FILTER", "WEEKEND_FILTER"),
("SESSION FILTER", "SESSION_FILTER"),
("DAY FILTER", "DAY_FILTER"),
("LOW ATR FILTER", "LOW_ATR_FILTER"),
("REGIME FILTER", "REGIME_FILTER"),
# Soft penalties / context warnings
("TREND EXHAUSTION", "TREND_EXHAUSTION"),
("EXHAUSTION", "EXHAUSTION_PENALTY"),
("NO PULLBACK", "NO_PULLBACK_PENALTY"),
("Weak pullback", "WEAK_PULLBACK_PENALTY"),
("Extended — RSI", "RSI_EXTENDED_PENALTY"),
("Counter-trend", "COUNTER_TREND_PENALTY"),
("EXTREME volatility", "EXTREME_VOL_PENALTY"),
("NEAR EMA200", "NEAR_EMA200_PENALTY"),
("FAR FROM EMA200", "FAR_EMA200_PENALTY"),
("EVENT RISK", "EVENT_RISK"),
("Timeframes CONFLICT", "MTF_CONFLICT"),
]
for w in report.warnings:
for tag, name in filter_tags:
if tag in w:
filters_hit.append(name)
break
# Determine outcome
non_neutral = report.final_signal not in (SignalStrength.NEUTRAL,)
quality_ok = report.signal_quality in ("A+", "A", "B", "C")
if quality_ok and report.trade_setup is not None:
return "TRADE", filters_hit
elif quality_ok and report.trade_setup is None:
return "QUALITY_OK_NO_SETUP", filters_hit
elif non_neutral and not quality_ok:
return "LOW_QUALITY", filters_hit
else:
return "NEUTRAL", filters_hit
# ═════════════════════════════════════════════════════════════════════
# SCORE DISTRIBUTION TRACKER
# ═════════════════════════════════════════════════════════════════════
class ScoreTracker:
"""Tracks score/confidence distributions for analysis."""
def __init__(self):
self.scores: List[float] = []
self.confidences: List[float] = []
self.abs_scores: List[float] = []
# How many bars had score in various ranges
self.score_bins = Counter()
self.conf_bins = Counter()
# Track "near miss" bars — score close to thresholds
self.near_miss_score: int = 0 # score 20-29 (almost BUY/SELL)
self.near_miss_conf: int = 0 # confidence 45-54 (almost passing)
def record(self, score: float, confidence: float):
self.scores.append(score)
self.confidences.append(confidence)
self.abs_scores.append(abs(score))
# Score bins
a = abs(score)
if a < 10:
self.score_bins["0-9"] += 1
elif a < 20:
self.score_bins["10-19"] += 1
elif a < 30:
self.score_bins["20-29"] += 1
self.near_miss_score += 1
elif a < 40:
self.score_bins["30-39"] += 1
elif a < 50:
self.score_bins["40-49"] += 1
else:
self.score_bins["50+"] += 1
# Confidence bins
if confidence < 30:
self.conf_bins["0-29"] += 1
elif confidence < 45:
self.conf_bins["30-44"] += 1
elif confidence < 55:
self.conf_bins["45-54"] += 1
self.near_miss_conf += 1
elif confidence < 65:
self.conf_bins["55-64"] += 1
elif confidence < 75:
self.conf_bins["65-74"] += 1
else:
self.conf_bins["75+"] += 1
# ═════════════════════════════════════════════════════════════════════
# MAIN DIAGNOSTIC LOOP
# ═════════════════════════════════════════════════════════════════════
def run_diagnosis(
pair: str,
start: datetime,
end: datetime,
sample_every: int = 4,
use_cache: bool = True,
primary_tf: Timeframe = Timeframe.M15,
confirm_tfs: Optional[List[Timeframe]] = None,
warmup_bars: int = 250,
) -> None:
if confirm_tfs is None:
confirm_tfs = [Timeframe.H1, Timeframe.H4]
pair_info = PAIR_INFO[pair]
display = pair_info["display"]
is_xau = display == "XAU/USD"
platform_cfg = PlatformConfig(
symbol=pair,
display_name=display,
primary_timeframe=primary_tf,
confirmation_timeframes=list(confirm_tfs),
)
provider = make_data_provider(platform_cfg)
# ── Load data ──
timeframes = [primary_tf] + list(confirm_tfs)
pad_days = max(60, warmup_bars // 4)
fetch_start = start - timedelta(days=pad_days)
dfs_full: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
df = fetch_history(
provider, pair, tf,
start=fetch_start, end=end,
use_cache=use_cache,
)
if df.empty:
raise RuntimeError(f"No data for {pair} {tf.value}")
dfs_full[tf] = df
# ── Pre-compute indicators ──
t0 = time.perf_counter()
caches: Dict[str, IndicatorCache] = {
tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True)
for tf in timeframes
}
logger.info(f"Indicator pre-compute: {time.perf_counter() - t0:.2f}s")
# ── Build aggregator with GPU engine ──
aggregator = SignalAggregator(platform_cfg)
aggregator.tech_engine = GPUTechnicalEngine(platform_cfg.technical, caches)
# ── Walk primary TF ──
primary_df = dfs_full[primary_tf]
bt_start_ts = pd.Timestamp(start)
bt_end_ts = pd.Timestamp(end)
first_eligible = max(
warmup_bars,
int(primary_df.index.searchsorted(bt_start_ts)),
)
last_pos = min(
int(primary_df.index.searchsorted(bt_end_ts, side="right")),
len(primary_df),
)
primary_dur = pd.Timedelta(seconds=BAR_SECONDS[primary_tf])
# Counters
total_bars_in_window = 0
total_bars_sampled = 0
outcome_counts = Counter()
filter_counts = Counter()
# Filters that killed bars that were otherwise directional (non-NEUTRAL score)
directional_filter_counts = Counter()
# Pre-pipeline rejections (XAU calendar filters applied before analyze_from_dataframes)
pre_pipeline = Counter()
# Track signal directions
signal_dir_counts = Counter()
quality_counts = Counter()
# Score tracker
tracker = ScoreTracker()
# Track "would-have-been" trades: bars where the primary score was
# directional but got killed by a specific filter
filter_kill_breakdown = defaultdict(Counter)
logger.info(
f"Diagnosing {display} {primary_tf.value}: "
f"bars {first_eligible}-{last_pos} (every {sample_every}th bar)"
)
t_loop = time.perf_counter()
for pos in range(first_eligible, last_pos, sample_every):
total_bars_in_window += 1
ts = primary_df.index[pos]
decision_time = ts + primary_dur
# ── XAU calendar pre-filter (same as gpu_backtest.py) ──
if is_xau:
block = _is_xau_blocked_by_calendar(ts, platform_cfg)
if block:
pre_pipeline[block] += 1
continue
total_bars_sampled += 1
# ── Build sliced dfs (same logic as gpu_backtest.py) ──
slice_dfs: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
full = dfs_full[tf]
bar_dur = pd.Timedelta(seconds=BAR_SECONDS[tf])
cutoff = decision_time - bar_dur
cut = int(full.index.searchsorted(cutoff, side="right"))
if cut == 0:
continue
slice_dfs[tf] = full.iloc[:cut]
if primary_tf not in slice_dfs:
continue
# ── Run the full signal pipeline ──
report: SignalReport = aggregator.analyze_from_dataframes(
symbol=pair,
dfs=slice_dfs,
context=None,
use_news=False,
)
# Record score/confidence
tracker.record(report.final_score, report.final_confidence)
# Signal direction
signal_dir_counts[report.final_signal.name] += 1
quality_counts[report.signal_quality] += 1
# Classify
outcome, filters_hit = classify_bar(report, is_xau)
outcome_counts[outcome] += 1
for f in filters_hit:
filter_counts[f] += 1
# If the primary score was directional (>= 15 or <= -15), the bar
# had some potential — track what killed it
if abs(report.primary_score) >= 15:
for f in filters_hit:
directional_filter_counts[f] += 1
# For bars that got quality D/F but had a directional signal,
# track which filter warnings appeared
if outcome in ("LOW_QUALITY", "QUALITY_OK_NO_SETUP"):
for f in filters_hit:
filter_kill_breakdown[outcome][f] += 1
elapsed = time.perf_counter() - t_loop
# ═════════════════════════════════════════════════════════════════
# PRINT RESULTS
# ═════════════════════════════════════════════════════════════════
bar = "=" * 80
print(f"\n{bar}")
print(f" FILTER REJECTION DIAGNOSTIC — {display}")
print(f" {start.date()} to {end.date()} | {primary_tf.value} "
f"(sampled every {sample_every}th bar)")
print(f"{bar}")
# Scale counts if sampling
scale_note = ""
if sample_every > 1:
scale_note = f" (sampled {total_bars_sampled} of ~{total_bars_in_window * sample_every} eligible bars)"
print(f"\n Total bars in backtest window: {total_bars_in_window}")
if pre_pipeline:
for reason, count in sorted(pre_pipeline.items()):
print(f" Pre-pipeline rejection ({reason}): {count}")
print(f" Bars analyzed: {total_bars_sampled}{scale_note}")
print(f" Analysis took: {elapsed:.1f}s "
f"({total_bars_sampled / max(elapsed, 0.001):.0f} bars/s)")
# ── Outcome breakdown ──
print(f"\n{'-' * 80}")
print(" OUTCOME BREAKDOWN")
print(f"{'-' * 80}")
outcome_order = ["TRADE", "QUALITY_OK_NO_SETUP", "LOW_QUALITY", "NEUTRAL"]
outcome_labels = {
"TRADE": "Actual trades (quality C+ & setup exists)",
"QUALITY_OK_NO_SETUP": "Quality C+ but NO trade setup generated",
"LOW_QUALITY": "Directional signal but quality D/F",
"NEUTRAL": "NEUTRAL (score didn't reach +/-30 or conf < 55)",
}
for o in outcome_order:
count = outcome_counts.get(o, 0)
pct = count / max(total_bars_sampled, 1) * 100
label = outcome_labels.get(o, o)
print(f" {count:6d} ({pct:5.1f}%) {label}")
# ── Signal direction breakdown ──
print(f"\n{'-' * 80}")
print(" SIGNAL DIRECTION BREAKDOWN")
print(f"{'-' * 80}")
for name in ["STRONG_BUY", "BUY", "NEUTRAL", "SELL", "STRONG_SELL"]:
count = signal_dir_counts.get(name, 0)
pct = count / max(total_bars_sampled, 1) * 100
print(f" {count:6d} ({pct:5.1f}%) {name}")
# ── Quality grade breakdown ──
print(f"\n{'-' * 80}")
print(" QUALITY GRADE BREAKDOWN")
print(f"{'-' * 80}")
for grade in ["A+", "A", "B", "C", "D", "F"]:
count = quality_counts.get(grade, 0)
pct = count / max(total_bars_sampled, 1) * 100
print(f" {count:6d} ({pct:5.1f}%) Grade {grade}")
# ── Filter hit counts (all bars) ──
print(f"\n{'-' * 80}")
print(" FILTER HIT COUNTS (all bars, a bar can hit multiple filters)")
print(f"{'-' * 80}")
if filter_counts:
for name, count in filter_counts.most_common():
pct = count / max(total_bars_sampled, 1) * 100
print(f" {count:6d} ({pct:5.1f}%) {name}")
else:
print(" (no filters triggered)")
# ── Filters that killed directional bars (primary_score >= 15) ──
print(f"\n{'-' * 80}")
print(" FILTERS KILLING DIRECTIONAL BARS (|primary_score| >= 15)")
print(f"{'-' * 80}")
if directional_filter_counts:
for name, count in directional_filter_counts.most_common():
print(f" {count:6d} {name}")
else:
print(" (none)")
# ── Filter breakdown for LOW_QUALITY bars ──
if filter_kill_breakdown.get("LOW_QUALITY"):
print(f"\n{'-' * 80}")
print(" FILTERS ON LOW_QUALITY BARS (had BUY/SELL but quality D/F)")
print(f"{'-' * 80}")
for name, count in filter_kill_breakdown["LOW_QUALITY"].most_common():
print(f" {count:6d} {name}")
# ── Filter breakdown for QUALITY_OK_NO_SETUP bars ──
if filter_kill_breakdown.get("QUALITY_OK_NO_SETUP"):
print(f"\n{'-' * 80}")
print(" FILTERS ON QUALITY_OK_NO_SETUP BARS (quality C+ but no trade setup)")
print(f"{'-' * 80}")
for name, count in filter_kill_breakdown["QUALITY_OK_NO_SETUP"].most_common():
print(f" {count:6d} {name}")
# ── Score distribution ──
print(f"\n{'-' * 80}")
print(" SCORE DISTRIBUTION (|score|)")
print(f"{'-' * 80}")
for bin_name in ["0-9", "10-19", "20-29", "30-39", "40-49", "50+"]:
count = tracker.score_bins.get(bin_name, 0)
pct = count / max(total_bars_sampled, 1) * 100
marker = " <-- near miss (almost BUY/SELL)" if bin_name == "20-29" else ""
print(f" {count:6d} ({pct:5.1f}%) |score| {bin_name}{marker}")
# ── Confidence distribution ──
print(f"\n{'-' * 80}")
print(" CONFIDENCE DISTRIBUTION")
print(f"{'-' * 80}")
for bin_name in ["0-29", "30-44", "45-54", "55-64", "65-74", "75+"]:
count = tracker.conf_bins.get(bin_name, 0)
pct = count / max(total_bars_sampled, 1) * 100
marker = " <-- near miss (almost passing)" if bin_name == "45-54" else ""
print(f" {count:6d} ({pct:5.1f}%) confidence {bin_name}{marker}")
# ── Near misses ──
print(f"\n{'-' * 80}")
print(" NEAR MISSES")
print(f"{'-' * 80}")
print(f" {tracker.near_miss_score:6d} Bars with |score| 20-29 (threshold is 30)")
print(f" {tracker.near_miss_conf:6d} Bars with confidence 45-54 (threshold is 55)")
# ── Summary statistics ──
if tracker.scores:
scores_arr = np.array(tracker.scores)
conf_arr = np.array(tracker.confidences)
print(f"\n{'-' * 80}")
print(" SCORE/CONFIDENCE STATISTICS")
print(f"{'-' * 80}")
print(f" Score: mean={np.mean(scores_arr):+.1f} "
f"median={np.median(scores_arr):+.1f} "
f"std={np.std(scores_arr):.1f} "
f"min={np.min(scores_arr):+.1f} "
f"max={np.max(scores_arr):+.1f}")
print(f" Confidence: mean={np.mean(conf_arr):.1f} "
f"median={np.median(conf_arr):.1f} "
f"std={np.std(conf_arr):.1f} "
f"min={np.min(conf_arr):.1f} "
f"max={np.max(conf_arr):.1f}")
# ── Blocked hours analysis (for XAU) ──
if is_xau:
blocked = set(platform_cfg.xau_strategy.blocked_hours_utc)
all_hours = set(range(24))
open_hours = sorted(all_hours - blocked)
print(f"\n{'-' * 80}")
print(" XAU HOUR/DAY CONFIG")
print(f"{'-' * 80}")
print(f" Blocked hours: {sorted(blocked)}")
print(f" Open hours: {open_hours}")
print(f" Allowed days: {platform_cfg.xau_strategy.allowed_weekdays} "
f"(0=Mon..4=Fri)")
print(f" Open fraction: {len(open_hours)/24:.0%} of hours x "
f"{len(platform_cfg.xau_strategy.allowed_weekdays)}/5 weekdays = "
f"{len(open_hours)/24 * len(platform_cfg.xau_strategy.allowed_weekdays)/5:.0%} of trading time")
# ── Interpretation / key takeaways ──
print(f"\n{bar}")
print(" KEY TAKEAWAYS")
print(f"{bar}")
neutral_pct = outcome_counts.get("NEUTRAL", 0) / max(total_bars_sampled, 1) * 100
low_q_pct = outcome_counts.get("LOW_QUALITY", 0) / max(total_bars_sampled, 1) * 100
trade_count = outcome_counts.get("TRADE", 0)
print(f" - {neutral_pct:.1f}% of bars never reach a directional signal (NEUTRAL)")
print(f" - {low_q_pct:.1f}% of bars get a BUY/SELL but confidence/quality too low")
print(f" - {trade_count} bars produce actual trades")
if filter_counts:
top3 = filter_counts.most_common(3)
print(f" - Top rejection filters:")
for name, count in top3:
print(f" {name}: {count} bars hit")
if tracker.near_miss_score > 50:
print(f" - SIGNIFICANT: {tracker.near_miss_score} bars had |score| 20-29 "
f"(lowering threshold from 30 to 25 would capture many)")
if tracker.near_miss_conf > 50:
print(f" - SIGNIFICANT: {tracker.near_miss_conf} bars had confidence 45-54 "
f"(lowering threshold from 55 to 50 would capture many)")
print(f"\n{bar}\n")
# ═════════════════════════════════════════════════════════════════════
# CLI
# ═════════════════════════════════════════════════════════════════════
def _resolve_pair(s: str) -> str:
s_up = s.upper().strip()
if s_up in PAIR_ALIASES:
return PAIR_ALIASES[s_up]
if s_up in PAIR_INFO:
return s_up
raise SystemExit(f"Unknown pair '{s}'. Valid: {', '.join(sorted(PAIR_ALIASES))}")
def _resolve_tf(s: str) -> Timeframe:
for tf in Timeframe:
if tf.value == s:
return tf
raise SystemExit(f"Unknown timeframe '{s}'")
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(
description="Diagnose which filters reject the most bars in the backtest pipeline",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p.add_argument("--pair", required=True, help="XAUUSD | GOLD")
p.add_argument("--days", type=int, default=365, help="Days of history")
p.add_argument("--start", help="ISO date YYYY-MM-DD (overrides --days)")
p.add_argument("--end", help="ISO date YYYY-MM-DD (defaults to today)")
p.add_argument("--primary-tf", default="15min", help="Primary TF")
p.add_argument("--confirm-tfs", default="1h,4h", help="Comma-separated confirmation TFs")
p.add_argument("--warmup-bars", type=int, default=250)
p.add_argument("--sample-every", type=int, default=4,
help="Analyze every Nth bar (1=all, 4=25%% sample for speed)")
p.add_argument("--no-cache", action="store_true", help="Skip parquet cache")
p.add_argument("--log-level", default="INFO")
args = p.parse_args(argv)
logging.basicConfig(
level=args.log_level.upper(),
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
pair = _resolve_pair(args.pair)
primary_tf = _resolve_tf(args.primary_tf)
confirm_tfs = [_resolve_tf(s.strip()) for s in args.confirm_tfs.split(",") if s.strip()]
end = (
datetime.fromisoformat(args.end).replace(tzinfo=timezone.utc)
if args.end else datetime.now(timezone.utc)
)
if args.start:
start = datetime.fromisoformat(args.start).replace(tzinfo=timezone.utc)
else:
start = end - timedelta(days=args.days)
logger.info(
f"Diagnosing {PAIR_INFO[pair]['display']} | "
f"Window: {start.date()} -> {end.date()} | "
f"Sample: every {args.sample_every}th bar"
)
run_diagnosis(
pair=pair,
start=start,
end=end,
sample_every=args.sample_every,
use_cache=not args.no_cache,
primary_tf=primary_tf,
confirm_tfs=confirm_tfs,
warmup_bars=args.warmup_bars,
)
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/backfill_news_archive.py
-230
View File
@@ -1,230 +0,0 @@
#!/usr/bin/env python3
"""
Seed logs/newsdata.db from the headline blobs still on disk.
==============================================================
The durable headline corpus (``news_archive.py``) only started collecting
when its hook shipped. Everything before that survives only inside signal
logs, in two different shapes:
* ``logs/signals.db`` — modern schema, headlines interned into
``news_snapshots`` (kind ``headlines``) and referenced by hash.
* an older orphan DB (default ``~/signals.db``) — pre-interning, blobs
inline in ``signals.headlines_json``, short-form item keys with no
``description``, ``link``, ``publisher`` or ``score_by_currency``.
Both are opened READ-ONLY. The script is idempotent: re-running it adds
nothing new and never downgrades a score.
Provenance is RECOMPUTED, not trusted
-------------------------------------
Neither source records which scorer produced its ``score`` — that flag
only exists from this change onwards. So each headline is re-scored with
the live keyword scorer and compared:
* stored score differs from the keyword score -> ``llm`` (DeepSeek wrote it)
* exact match on a non-zero score -> ``keyword``
* both zero -> ``unknown`` (indistinguishable)
Rows are stamped ``origin='backfill'`` because they come from the
post-filter pool: ``get_fundamental_context`` drops headlines inside
±0.1 before they ever reach a signal row, so genuinely neutral headlines
are under-represented in this history. Rows collected live by the hook
in ``news_engine`` do not have that bias and are stamped ``'live'``.
Usage
-----
venv/bin/python scripts/backfill_news_archive.py --dry-run
venv/bin/python scripts/backfill_news_archive.py
venv/bin/python scripts/backfill_news_archive.py --legacy-db /path/to/old.db
"""
from __future__ import annotations
import argparse
import json
import os
import sqlite3
import sys
from collections import Counter
from typing import Any, Dict, Iterator, Tuple
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import app_config # noqa: E402
import news_archive # noqa: E402
from config import NewsConfig # noqa: E402
from news_engine import NewsEngine # noqa: E402
def _open_ro(path: str) -> sqlite3.Connection:
"""Read-only, but NOT immutable=1 — that flag is blind to WAL rows."""
return sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True)
def _has_table(con: sqlite3.Connection, name: str) -> bool:
row = con.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
(name,)).fetchone()
return row is not None
def iter_blobs(path: str) -> Iterator[Tuple[str, str]]:
"""Yield (headlines_json, seen_at) from either DB shape."""
con = _open_ro(path)
try:
if _has_table(con, "news_snapshots"):
yield from con.execute(
"SELECT body, first_seen FROM news_snapshots"
" WHERE kind='headlines' ORDER BY first_seen")
cols = {r[1] for r in con.execute("PRAGMA table_info(signals)")}
if "headlines_json" in cols:
order = "logged_at" if "logged_at" in cols else "id"
yield from con.execute(
"SELECT headlines_json, "
+ ("logged_at" if "logged_at" in cols else "''")
+ " FROM signals WHERE headlines_json IS NOT NULL"
f" ORDER BY {order}")
finally:
con.close()
class _Row:
"""Duck-typed stand-in for NewsSentiment, so news_archive.build_row
flattens a stored dict exactly the way it flattens a live headline."""
__slots__ = ("headline", "source", "timestamp", "currencies_mentioned",
"sentiment_score", "keywords", "sentiment_by_currency",
"link", "description", "publisher", "score_source")
def __init__(self, item: Dict[str, Any], score_source: str):
self.headline = item.get("headline")
self.source = item.get("source")
self.timestamp = item.get("timestamp")
self.currencies_mentioned = item.get("currencies") or []
self.sentiment_score = item.get("score")
self.keywords = item.get("keywords") or []
self.sentiment_by_currency = item.get("score_by_currency") or {}
self.link = item.get("link")
self.description = item.get("description")
self.publisher = item.get("publisher")
self.score_source = score_source
def classify(engine: NewsEngine, item: Dict[str, Any]) -> str:
"""Recover which scorer produced this stored score."""
text = item.get("headline")
if not isinstance(text, str) or not text.strip():
return "unknown"
try:
stored = float(item.get("score") or 0.0)
except (TypeError, ValueError):
return "unknown"
keyword = engine._analyze_headline_sentiment(text)
if abs(stored - keyword) > 1e-9:
return "llm"
if abs(stored) < 1e-9:
return "unknown" # both zero — the two scorers are indistinguishable
return "keyword"
def collect(paths, engine: NewsEngine):
"""Newest wins per headline: later blobs carry upgraded LLM scores."""
best: Dict[str, Tuple[str, Dict[str, Any], str]] = {}
stats = Counter()
for path in paths:
if not os.path.exists(path):
print(f" skip (not found): {path}")
continue
blobs = 0
for body, seen_at in iter_blobs(path):
try:
items = json.loads(body)
except Exception:
stats["bad_blob"] += 1
continue
blobs += 1
for item in items:
if not isinstance(item, dict):
continue
text = item.get("headline")
if not isinstance(text, str) or not text.strip():
continue
source = classify(engine, item)
prior = best.get(text)
if prior is None or (
news_archive._SOURCE_RANK.get(source, 0)
>= news_archive._SOURCE_RANK.get(prior[0], 0)):
best[text] = (source, item, seen_at or "")
stats["items"] += 1
print(f" {path}: {blobs} blobs")
for source, _, _ in best.values():
stats[source] += 1
return best, stats
def main() -> int:
default_legacy = os.path.join(os.path.expanduser("~"), "signals.db")
ap = argparse.ArgumentParser(description=__doc__.split("\n")[1])
ap.add_argument("--db", default=app_config.signal_log_path(),
help="modern signal-log DB (default: config path)")
ap.add_argument("--legacy-db", default=default_legacy,
help=f"older inline-blob DB (default: {default_legacy})")
ap.add_argument("--out", default=app_config.news_archive_path(),
help="archive DB to write (default: config path)")
ap.add_argument("--dry-run", action="store_true",
help="count only, write nothing")
args = ap.parse_args()
engine = NewsEngine(NewsConfig())
print("Reading headline blobs (read-only):")
best, stats = collect([args.db, args.legacy_db], engine)
print(f"\n headline items scanned : {stats['items']}")
print(f" distinct headlines : {len(best)}")
print(f" llm (DeepSeek) : {stats['llm']}")
print(f" keyword : {stats['keyword']}")
print(f" unknown (both zero) : {stats['unknown']}")
if stats["bad_blob"]:
print(f" unparseable blobs : {stats['bad_blob']}")
if args.dry_run:
print("\n--dry-run: nothing written.")
return 0
if not best:
print("\nNothing to write.")
return 0
rows = []
for text, (source, item, seen_at) in best.items():
row = news_archive.build_row(
_Row(item, source),
teacher_model=None, # not recorded by either source
teacher_prompt_sha1=None,
origin="backfill",
first_seen=seen_at or None,
last_seen=seen_at or None)
if row is not None:
rows.append(row)
archive = news_archive.NewsArchive(args.out)
try:
written = archive.store(rows)
finally:
archive.close()
print(f"\nWrote {written} rows to {args.out}")
con = _open_ro(args.out)
try:
print("\nArchive now holds:")
for source, count in con.execute(
"SELECT score_source, COUNT(*) FROM headlines"
" GROUP BY score_source ORDER BY COUNT(*) DESC"):
print(f" {source:8s} {count}")
finally:
con.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/migrate_signal_log_news.py
-177
View File
@@ -1,177 +0,0 @@
#!/usr/bin/env python3
"""
migrate_signal_log_news.py — move inline news blobs into news_snapshots.
Legacy rows store the full RSS headline list (~33 KB) and calendar events
(~1.2 KB) inline on EVERY scan row; the same blob repeats ~5× because the
news caches outlive a candle. This script interns each distinct blob once
(content hash), stamps `headlines_hash` / `events_hash` on the row and
NULLs the inline columns. Readers use the `signals_with_news` view or
`signal_log.read_news`.
Idempotent and resumable (batched transactions). Run --vacuum with BOTH
live writers (telegram bot, dashboard publisher) stopped — VACUUM needs
exclusive access and ~1× the final size of free disk.
python scripts/migrate_signal_log_news.py --dry-run
python scripts/migrate_signal_log_news.py
python scripts/migrate_signal_log_news.py --vacuum # bots stopped
"""
import argparse
import hashlib
import os
import sqlite3
import sys
from datetime import datetime, timezone
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import app_config # noqa: E402
from signal_log import SignalLogger # noqa: E402
def _hash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def migrate(con: sqlite3.Connection, batch: int = 500) -> dict:
"""Intern legacy inline blobs; returns counters. Caller owns the
connection (schema must already exist — open a SignalLogger once)."""
stats = {"rows": 0, "headlines_interned": 0, "events_interned": 0}
now = datetime.now(timezone.utc).isoformat()
while True:
rows = con.execute(
"SELECT id, headlines_json, events_json FROM signals"
" WHERE (headlines_json IS NOT NULL AND headlines_hash IS NULL)"
" OR (events_json IS NOT NULL AND events_hash IS NULL)"
" ORDER BY id LIMIT ?", (batch,)).fetchall()
if not rows:
break
con.execute("BEGIN IMMEDIATE")
try:
for rowid, headlines, events in rows:
sets, params = [], []
if headlines is not None:
h = _hash(headlines)
cur = con.execute(
"INSERT OR IGNORE INTO news_snapshots"
" (hash, kind, body, first_seen, last_seen)"
" VALUES (?, 'headlines', ?, ?, ?)",
(h, headlines, now, now))
stats["headlines_interned"] += cur.rowcount
sets += ["headlines_hash=?", "headlines_json=NULL"]
params.append(h)
if events is not None:
h = _hash(events)
cur = con.execute(
"INSERT OR IGNORE INTO news_snapshots"
" (hash, kind, body, first_seen, last_seen)"
" VALUES (?, 'events', ?, ?, ?)",
(h, events, now, now))
stats["events_interned"] += cur.rowcount
sets += ["events_hash=?", "events_json=NULL"]
params.append(h)
params.append(rowid)
con.execute(
f"UPDATE signals SET {', '.join(sets)} WHERE id=?",
params)
stats["rows"] += 1
con.commit()
except Exception:
con.rollback()
raise
return stats
def prune_orphans(con: sqlite3.Connection) -> int:
cur = con.execute(
"DELETE FROM news_snapshots WHERE hash NOT IN"
" (SELECT headlines_hash FROM signals WHERE headlines_hash IS NOT NULL"
" UNION SELECT events_hash FROM signals WHERE events_hash IS NOT NULL)")
con.commit()
return cur.rowcount
def prune_news_older_than(con: sqlite3.Connection, days: int) -> int:
"""Drop the news reference (not the row) from rows older than N days,
then delete snapshots nothing references any more."""
# logged_at is ISO8601 with a 'T' and a +00:00 offset; SQLite's
# datetime() returns 'YYYY-MM-DD HH:MM:SS'. Comparing them as raw
# strings put every row from the cutoff CALENDAR DAY on the wrong
# side ('T' > ' '), so a full day never pruned. Normalise to the
# same shape first — all logged_at values are UTC (see _iso()).
cur = con.execute(
"UPDATE signals SET headlines_hash=NULL, events_hash=NULL"
" WHERE substr(replace(logged_at, 'T', ' '), 1, 19)"
" < datetime('now', ?)", (f"-{int(days)} days",))
con.commit()
prune_orphans(con)
return cur.rowcount
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[1])
ap.add_argument("--db", default=None)
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--batch", type=int, default=500)
ap.add_argument("--prune-news-older-than", type=int, default=None,
metavar="DAYS")
ap.add_argument("--prune-orphans", action="store_true")
ap.add_argument("--vacuum", action="store_true",
help="checkpoint + VACUUM (all writers must be stopped)")
ap.add_argument("--no-backup", action="store_true")
args = ap.parse_args(argv)
path = args.db or app_config.signal_log_path()
if not os.path.exists(path):
print(f"no database at {path}")
return 1
size0 = os.path.getsize(path)
if args.dry_run:
con = sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True)
n = con.execute(
"SELECT COUNT(*) FROM signals WHERE headlines_json IS NOT NULL"
" OR events_json IS NOT NULL").fetchone()[0]
d = con.execute(
"SELECT COUNT(DISTINCT headlines_json) FROM signals"
" WHERE headlines_json IS NOT NULL").fetchone()[0]
print(f"{n} rows carry inline news blobs ({d} distinct headline "
f"blobs); file size {size0 / 1e6:.1f} MB")
con.close()
return 0
if not args.no_backup:
backup = f"{path}.pre-news-{datetime.now(timezone.utc):%Y%m%d}"
if not os.path.exists(backup):
src = sqlite3.connect(path)
dst = sqlite3.connect(backup)
src.backup(dst)
dst.close()
src.close()
print(f"backup written: {backup}")
lg = SignalLogger(path) # applies the additive schema
lg.close()
con = sqlite3.connect(path)
con.execute("PRAGMA busy_timeout=10000")
try:
print("migrate:", migrate(con, batch=args.batch))
if args.prune_news_older_than is not None:
print("rows de-referenced:",
prune_news_older_than(con, args.prune_news_older_than))
if args.prune_orphans:
print("orphan snapshots removed:", prune_orphans(con))
if args.vacuum:
try:
con.execute("PRAGMA wal_checkpoint(TRUNCATE)")
con.execute("VACUUM")
except sqlite3.OperationalError as exc:
print(f"VACUUM failed ({exc}) — stop the bot and dashboard "
"publisher first, then re-run with --vacuum")
return 2
finally:
con.close()
print(f"size {size0 / 1e6:.1f} MB -> {os.path.getsize(path) / 1e6:.1f} MB"
+ ("" if args.vacuum else " (run --vacuum to reclaim space)"))
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/shadow_scan.py
-192
View File
@@ -1,192 +0,0 @@
#!/usr/bin/env python3
"""
Phase 4 — live shadow scan (data source change test)
=====================================================
Runs TWO data providers side-by-side at every candle close (M5/M15/H1)
and logs how their signals diverge, so the feed switch can be validated
under real fetch depths / caching / boundary timing for ≥2 weeks.
Unlike the history studies, this uses the LIVE fetch path (fetch_ohlcv,
forming-candle drop, freshness checks) for both providers, but calls
`analyze_from_dataframes(use_news=False)` directly so the (provider-
independent) news/DeepSeek budget is not burned on shadow scans. The
weekend filter is applied here (skip the whole cycle), and the session/
day/hour gates run inside analyze_from_dataframes as in the backtest.
No orders, no Telegram, no signal-log writes — output is a side-by-side
CSV plus a rolling agreement tally.
Usage:
venv/bin/python scripts/shadow_scan.py --a oanda --b tiingo \
--tfs M5,M15,H1 --out logs/shadow_oanda_tiingo.csv --once
# production (2+ weeks):
nohup venv/bin/python scripts/shadow_scan.py --a oanda --b tiingo \
--out logs/shadow_oanda_tiingo.csv >> logs/shadow_scan.log 2>&1 &
CSV columns (one row per TF per boundary):
ts, tf, a_<field>, b_<field> for direction/grade/setup/tag/entry/sl/tp1,
a_price, b_price, a_spread, b_spread, a_warnings, b_warnings,
dir_agree, grade_agree, setup_agree
"""
import argparse
import csv
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from config import PlatformConfig, Timeframe, XAUUSD
from pair_signals.xauusd import make_config
from providers import make_data_provider
from providers.base import _is_forex_weekend
from signal_aggregator import SignalAggregator, SignalReport
_TF_LOOKUP = {"M5": Timeframe.M5, "M15": Timeframe.M15, "H1": Timeframe.H1,
"H4": Timeframe.H4}
_TF_MINUTES = {Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.H1: 60,
Timeframe.H4: 240}
_CONFIRMS = {Timeframe.M5: [Timeframe.M15, Timeframe.H1],
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
Timeframe.H4: [Timeframe.H1, Timeframe.D1]}
FIELDS = ["direction", "grade", "setup", "tag", "entry", "sl", "tp1"]
def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
now = datetime.now(timezone.utc)
elapsed = (now.minute % interval_minutes) * 60 + now.second \
+ now.microsecond / 1e6
return interval_minutes * 60 - elapsed + buffer
def due_timeframes(wake: datetime, tfs) -> list:
minutes = wake.hour * 60 + wake.minute
floored = minutes - (minutes % 5)
return [tf for tf in tfs if floored % _TF_MINUTES[tf] == 0]
def tag_of(report) -> str:
fb = report.bullish_reasons[0] if report.bullish_reasons else ""
fbe = report.bearish_reasons[0] if report.bearish_reasons else ""
if "XAU PULLBACK" in fb or "XAU PULLBACK" in fbe:
return "XAU_PB"
if "XAU CHART PATTERN" in fb or "XAU CHART PATTERN" in fbe:
return "XAU_CP"
if "XAU STRATEGY" in fb:
return "XAU"
return "GENERIC"
def scan(provider, symbol, tf) -> SignalReport:
confirms = _CONFIRMS[tf]
cfg = make_config(tf, confirms)
agg = SignalAggregator(cfg)
agg.data_provider = provider
dfs = {}
for t in [tf] + confirms:
dfs[t] = provider.fetch_ohlcv(symbol, t)
return agg.analyze_from_dataframes(symbol, dfs, use_news=False)
def extract(provider, report) -> dict:
s = report.trade_setup
return {
"direction": report.final_signal.value,
"grade": report.signal_quality,
"setup": 1 if s is not None else 0,
"tag": tag_of(report) if s else "",
"entry": round(s.entry_price, 3) if s else "",
"sl": round(s.stop_loss, 3) if s else "",
"tp1": round(s.take_profit_1, 3) if s else "",
"price": round(report.current_price, 3),
"spread": round(provider.get_spread(XAUUSD), 3),
"warnings": " | ".join(report.warnings)[:200],
}
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--a", default="oanda")
ap.add_argument("--b", default="tiingo")
ap.add_argument("--tfs", default="M5,M15,H1")
ap.add_argument("--out", default="logs/shadow_scan.csv")
ap.add_argument("--once", action="store_true",
help="one immediate scan cycle, then exit")
args = ap.parse_args()
tfs = [_TF_LOOKUP[s.strip()] for s in args.tfs.split(",")]
pa = make_data_provider(PlatformConfig(), args.a)
pb = make_data_provider(PlatformConfig(), args.b)
out_path = Path(args.out)
out_path.parent.mkdir(parents=True, exist_ok=True)
new_file = not out_path.exists()
tally = {"cycles": 0, "dir": 0, "grade": 0, "setup": 0}
with open(out_path, "a", newline="") as fh:
w = csv.writer(fh)
if new_file:
w.writerow(["ts", "tf"] + [f"a_{x}" for x in FIELDS + ["price", "spread", "warnings"]]
+ [f"b_{x}" for x in FIELDS + ["price", "spread", "warnings"]]
+ ["dir_agree", "grade_agree", "setup_agree"])
while True:
if args.once:
wake = datetime.now(timezone.utc)
else:
wait = seconds_until_boundary(5, buffer=10)
time.sleep(wait)
wake = datetime.now(timezone.utc)
if _is_forex_weekend():
if args.once:
print("weekend — nothing to scan")
return
continue
for tf in due_timeframes(wake, tfs):
try:
ra = scan(pa, XAUUSD, tf)
except Exception as e:
print(f"[{wake:%H:%M:%S}] {args.a} {tf.value} FAILED: {e}")
continue
try:
rb = scan(pb, XAUUSD, tf)
except Exception as e:
print(f"[{wake:%H:%M:%S}] {args.b} {tf.value} FAILED: {e}")
continue
ea, eb = extract(pa, ra), extract(pb, rb)
da = ea["direction"] == eb["direction"]
ga = ea["grade"] == eb["grade"]
sa = ea["setup"] == eb["setup"]
tally["cycles"] += 1
tally["dir"] += da; tally["grade"] += ga; tally["setup"] += sa
row = [wake.isoformat(), tf.value]
row += [ea[x] for x in FIELDS + ["price", "spread", "warnings"]]
row += [eb[x] for x in FIELDS + ["price", "spread", "warnings"]]
row += [int(da), int(ga), int(sa)]
w.writerow(row)
fh.flush()
print(f"[{wake:%Y-%m-%d %H:%M:%S}] {tf.value:<5} "
f"{args.a}={ea['direction']:+d}/{ea['grade'] or '-':>2} "
f"{args.b}={eb['direction']:+d}/{eb['grade'] or '-':>2} "
f"dir={'OK' if da else 'FLIP'} "
f"| Δprice {abs(ea['price']-eb['price']):.2f} "
f"| agree dir {tally['dir']}/{tally['cycles']} "
f"grade {tally['grade']}/{tally['cycles']} "
f"setup {tally['setup']}/{tally['cycles']}")
if args.once:
print("shadow scan complete (--once)")
return
if __name__ == "__main__":
main()
test_all.py
+2 -92
View File
@@ -4544,10 +4544,10 @@ def _():
# ══════════════════════════════════════════════════════════════════════
# MODULE 18: --set research override + diagnose_filters tag freshness
# MODULE 18: --set research override
# ══════════════════════════════════════════════════════════════════════
print("\n═══ Module 18: --set override + diagnose tags ═══")
print("\n═══ Module 18: --set override ═══")
@test("--set: nested float/bool/JSON-list overrides apply with typing")
@@ -4603,42 +4603,6 @@ def _():
assert _parse_args(["--pair", "XAUUSD"]).set == []
@test("diagnose_filters: every filter tag has a live emitter (staleness guard)")
def _():
import inspect
import diagnose_filters
import signal_aggregator as sa
src = inspect.getsource(sa)
tag_src = inspect.getsource(diagnose_filters.classify_bar)
# Extract the tag strings from the ordered list in classify_bar.
import re
tags = re.findall(r'\("([^"]+)",\s+"[A-Z0-9_]+"\)', tag_src)
assert len(tags) >= 25, f"tag extraction broke: found {len(tags)}"
for tag in tags:
assert tag in src, f"filter tag '{tag}' has no emitter in signal_aggregator.py"
@test("diagnose_filters: one tag per warning, specific prefix wins")
def _():
import diagnose_filters as d
from signal_aggregator import SignalReport
class _R:
pass
r = _R()
r.warnings = [
"M5 FILTER: SELL quality A capped — high-conviction M5 shorts enter at extremes",
"SELL FILTER: 15:00 UTC entry blocked for shorts",
"TREND EXHAUSTION — price extended >3.5 ATR past swing high",
]
r.final_signal = SignalStrength.SELL
r.signal_quality = "D"
r.trade_setup = None
outcome, hits = d.classify_bar(r, True)
assert outcome == "LOW_QUALITY"
assert hits == ["M5_SELL_QUALITY", "SELL_HOUR", "TREND_EXHAUSTION"], hits
# ══════════════════════════════════════════════════════════════════════
# MODULE 19: DeepSeek news enrichment (llm_client + news_engine seams)
# ══════════════════════════════════════════════════════════════════════
@@ -8827,7 +8791,6 @@ def _():
def _():
import sqlite3
import signal_log as slog
from scripts.migrate_signal_log_news import migrate, prune_orphans
fd, path = tempfile.mkstemp(suffix=".db")
os.close(fd)
try:
@@ -8859,19 +8822,6 @@ def _():
"SELECT id, headlines_resolved FROM signals_with_news").fetchall())
assert json.loads(view[legacy_id])[0]["headline"] == "legacy"
assert json.loads(view[r2])[0]["headline"] == "Fed signals rate hike"
# migration moves the legacy blob and is idempotent
stats = migrate(lg._conn)
assert stats["rows"] == 1 and stats["headlines_interned"] == 1
assert migrate(lg._conn)["rows"] == 0
row = lg._conn.execute(
"SELECT headlines_json, headlines_hash FROM signals WHERE id=?",
(legacy_id,)).fetchone()
assert row[0] is None and row[1]
assert json.loads(slog.read_news(lg._conn, legacy_id)[1])[0][
"headline"] == "legacy"
lg._conn.execute("DELETE FROM signals WHERE id=?", (legacy_id,))
lg._conn.commit()
assert prune_orphans(lg._conn) == 2, "legacy headline + event blobs"
lg.close()
finally:
slog.reset_loggers_for_tests()
@@ -9127,46 +9077,6 @@ def _():
"a real evaluator fault must be visible in the log"
@test("migrate: prune_news_older_than prunes the cutoff calendar day")
def _():
import sqlite3
import importlib.util
spec = importlib.util.spec_from_file_location(
"_mig_test", os.path.join(os.path.dirname(os.path.abspath(__file__)),
"scripts", "migrate_signal_log_news.py"))
mig = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mig)
fd, path = tempfile.mkstemp(suffix=".db")
os.close(fd)
try:
con = sqlite3.connect(path)
con.execute("CREATE TABLE signals (id INTEGER PRIMARY KEY,"
" logged_at TEXT, headlines_hash TEXT, events_hash TEXT)")
con.execute("CREATE TABLE news_snapshots (hash TEXT PRIMARY KEY,"
" body TEXT)")
now = datetime.now(timezone.utc)
# Genuinely older than the 30-day cutoff, but on the SAME calendar
# day as it — the case the raw string compare put on the wrong side
# ('T' sorts after ' ').
boundary = (now - timedelta(days=30)).replace(
hour=0, minute=0, second=1, microsecond=0).isoformat()
con.executemany(
"INSERT INTO signals VALUES (?,?,?,?)",
[(1, boundary, "h1", "e1"), (2, now.isoformat(), "h2", "e2")])
con.executemany("INSERT INTO news_snapshots VALUES (?,?)",
[("h1", "{}"), ("e1", "{}"), ("h2", "{}"), ("e2", "{}")])
con.commit()
n = mig.prune_news_older_than(con, 30)
assert n == 1, f"the cutoff-day row must prune, got {n}"
kept = con.execute(
"SELECT headlines_hash FROM signals ORDER BY id").fetchall()
assert kept == [(None,), ("h2",)], kept
con.close()
finally:
os.unlink(path)
# ══════════════════════════════════════════════════════════════════════
# MODULE 29: stake-multiplier sizing — features + record-keeping
# ══════════════════════════════════════════════════════════════════════