Compare commits
3 Commits
2836024e9d
...
552199857a
| Author | SHA1 | Date | |
|---|---|---|---|
 Harwinder Sekhon
|
552199857a | ||
|
Harwinder Sekhon
|
b0b71e1225 | ||
|
Harwinder Sekhon
|
931b26548e |
@@ -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/
|
||||
|
||||
+12
-35
@@ -84,7 +84,12 @@ _VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
|
||||
_SCANNABLE_TFS = ("M5", "M15", "H1", "H4")
|
||||
|
||||
_KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level",
|
||||
"oanda", "deepseek", "agent_panel", "market_outlook",
|
||||
"oanda", "deepseek", "agent_panel",
|
||||
# "market_outlook" is TOLERATED but ignored: the AI market
|
||||
# outlook was removed 2026-09-15 (last commit 2836024). Live
|
||||
# config.json files still carry the block and must keep
|
||||
# loading; load_app_config logs one warning when it is there.
|
||||
"market_outlook",
|
||||
"violent_watch", "signal_log", "news_archive"}
|
||||
_KNOWN_OANDA = {
|
||||
# "account_id" is TOLERATED but ignored: it belonged to the auto-
|
||||
@@ -102,7 +107,6 @@ _KNOWN_AGENT_PANEL = {
|
||||
"candles_enabled", "candle_bars_primary", "candle_bars_confirm",
|
||||
"verdict_source",
|
||||
}
|
||||
_KNOWN_MARKET_OUTLOOK = {"enabled", "telegram", "max_per_day", "candles_bars"}
|
||||
_KNOWN_VIOLENT_WATCH = {"enabled", "alert_line", "headsup",
|
||||
"max_headsup_per_day", "headsup_min_level"}
|
||||
_KNOWN_SIGNAL_LOG = {"enabled", "path", "log_blocked_scans"}
|
||||
@@ -111,13 +115,6 @@ _KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"}
|
||||
_KNOWN_SCANNING = {"pairs", "timeframes"}
|
||||
_KNOWN_PAIR_KEYS = {"blocked_hours_utc"}
|
||||
|
||||
_MARKET_OUTLOOK_DEFAULTS = {
|
||||
"enabled": False,
|
||||
"telegram": False,
|
||||
"max_per_day": 12,
|
||||
"candles_bars": 20,
|
||||
}
|
||||
|
||||
# Violent-move watch (display-only heads-up) — dark by default.
|
||||
_VIOLENT_WATCH_DEFAULTS = {
|
||||
"enabled": False,
|
||||
@@ -149,8 +146,6 @@ class AppConfig:
|
||||
oanda_settings: Dict[str, Any] = field(default_factory=dict)
|
||||
deepseek_settings: Dict[str, Any] = field(default_factory=dict)
|
||||
agent_panel_settings: Dict[str, Any] = field(default_factory=dict)
|
||||
market_outlook_settings: Dict[str, Any] = field(
|
||||
default_factory=lambda: dict(_MARKET_OUTLOOK_DEFAULTS))
|
||||
violent_watch_settings: Dict[str, Any] = field(
|
||||
default_factory=lambda: dict(_VIOLENT_WATCH_DEFAULTS))
|
||||
signal_log_settings: Dict[str, Any] = field(default_factory=dict)
|
||||
@@ -341,19 +336,12 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig:
|
||||
'"chair"')
|
||||
app.agent_panel_settings = dict(panel)
|
||||
|
||||
outlook = raw.get("market_outlook", {})
|
||||
_check_keys(outlook, _KNOWN_MARKET_OUTLOOK, '"market_outlook"')
|
||||
for flag in ("enabled", "telegram"):
|
||||
if flag in outlook and not isinstance(outlook[flag], bool):
|
||||
raise AppConfigError(
|
||||
f"config.json: market_outlook.{flag} must be true or false")
|
||||
for key in ("max_per_day", "candles_bars"):
|
||||
if key in outlook:
|
||||
value = outlook[key]
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
|
||||
raise AppConfigError(
|
||||
f"config.json: market_outlook.{key} must be a positive integer")
|
||||
app.market_outlook_settings.update(outlook)
|
||||
if "market_outlook" in raw:
|
||||
# The AI market-outlook feature was removed 2026-09-15; older
|
||||
# config.json files still carry its block and must keep loading.
|
||||
logger.warning(
|
||||
"config.json: market_outlook block is ignored (feature removed "
|
||||
"2026-09-15) — delete it")
|
||||
|
||||
watch = raw.get("violent_watch", {})
|
||||
_check_keys(watch, _KNOWN_VIOLENT_WATCH, '"violent_watch"')
|
||||
@@ -493,17 +481,6 @@ def deepseek_api_key(app: Optional[AppConfig] = None) -> str:
|
||||
app.deepseek_settings.get("api_key", ""))
|
||||
|
||||
|
||||
# ── AI market-outlook resolution helper ──────────────────────────────
|
||||
|
||||
def market_outlook_enabled(app: Optional[AppConfig] = None) -> bool:
|
||||
"""Whether the advisory outlook feature is enabled (env wins)."""
|
||||
env = os.environ.get("MARKET_OUTLOOK_ENABLED")
|
||||
if env:
|
||||
return env.strip().lower() in ("1", "true", "yes", "on")
|
||||
app = app or get_app_config()
|
||||
return bool(app.market_outlook_settings.get("enabled", False))
|
||||
|
||||
|
||||
# ── Violent-move watch resolution helper ─────────────────────────────
|
||||
|
||||
def violent_watch_enabled(app: Optional[AppConfig] = None) -> bool:
|
||||
|
||||
@@ -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())
|
||||
@@ -48,12 +48,6 @@
|
||||
"candle_bars_confirm": 10,
|
||||
"verdict_source": "macro"
|
||||
},
|
||||
"market_outlook": {
|
||||
"enabled": false,
|
||||
"telegram": false,
|
||||
"max_per_day": 12,
|
||||
"candles_bars": 20
|
||||
},
|
||||
"violent_watch": {
|
||||
"enabled": false,
|
||||
"alert_line": true,
|
||||
|
||||
@@ -13,7 +13,6 @@ import json
|
||||
import logging
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
# Force UTF-8 output on Windows so box-drawing chars render
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
@@ -24,7 +23,6 @@ if hasattr(sys.stderr, "reconfigure"):
|
||||
from agent_panel import attach_panel
|
||||
from app_config import get_app_config, violent_watch_enabled
|
||||
from config import PlatformConfig, Timeframe, SignalStrength, XAUUSD
|
||||
from market_outlook import OutlookResult, maybe_run_outlook
|
||||
from pair_signals import PAIR_PRODUCERS
|
||||
from signal_aggregator import SignalAggregator, SignalReport
|
||||
from violent_watch import attach_watch
|
||||
@@ -205,24 +203,6 @@ def print_signal_report(report: SignalReport, detail: bool = False):
|
||||
print(f" {label:>20} {color}{val:.3f}{Colors.RESET}")
|
||||
|
||||
|
||||
def print_market_outlook(outlook: Optional[OutlookResult]) -> None:
|
||||
"""Render the independent advisory outlook for explicit CLI use."""
|
||||
print(f"\n {Colors.BOLD}AI Market Outlook{Colors.RESET}")
|
||||
if outlook is None:
|
||||
print(f" {Colors.GRAY}Unavailable (disabled, market closed, or no LLM response).{Colors.RESET}")
|
||||
return
|
||||
forming = {
|
||||
"buy": "possible buy forming",
|
||||
"sell": "possible sell forming",
|
||||
"none": "no clear setup forming",
|
||||
}[outlook.setup_forming]
|
||||
print(f" {outlook.bias.title()} · {forming} · {outlook.confidence:.0f}%")
|
||||
if outlook.watch:
|
||||
print(f" Watch: {outlook.watch}")
|
||||
if outlook.note:
|
||||
print(f" {Colors.GRAY}{outlook.note}{Colors.RESET}")
|
||||
|
||||
|
||||
def report_to_dict(report: SignalReport) -> dict:
|
||||
d = {
|
||||
"symbol": report.symbol,
|
||||
@@ -315,9 +295,6 @@ Examples:
|
||||
parser.add_argument("--panel", action="store_true",
|
||||
help="Run the LLM agent panel on the result "
|
||||
"(advisory only; needs a DeepSeek key)")
|
||||
parser.add_argument("--outlook", action="store_true",
|
||||
help="Run the hourly AI market outlook now "
|
||||
"(advisory only; needs a DeepSeek key)")
|
||||
parser.add_argument("--json", action="store_true",
|
||||
help="Output as JSON")
|
||||
parser.add_argument("--verbose", "-v", action="store_true",
|
||||
@@ -372,30 +349,13 @@ def run_live(args, config: PlatformConfig):
|
||||
attach_watch(report, config.primary_timeframe,
|
||||
get_app_config().violent_watch_settings)
|
||||
|
||||
outlook = None
|
||||
if getattr(args, "outlook", False):
|
||||
outlook = maybe_run_outlook(surface="cli", force=True)
|
||||
|
||||
if args.json:
|
||||
payload = report_to_dict(report)
|
||||
if getattr(args, "outlook", False):
|
||||
payload["market_outlook"] = (
|
||||
None if outlook is None else {
|
||||
"bias": outlook.bias,
|
||||
"setup_forming": outlook.setup_forming,
|
||||
"confidence": outlook.confidence,
|
||||
"watch": outlook.watch,
|
||||
"note": outlook.note,
|
||||
"created_ts": outlook.created_ts,
|
||||
}
|
||||
)
|
||||
print(json.dumps(payload, indent=2))
|
||||
return
|
||||
|
||||
print_header(f"{config.display_name} SIGNAL INTELLIGENCE")
|
||||
print_signal_report(report, detail=args.detail)
|
||||
if getattr(args, "outlook", False):
|
||||
print_market_outlook(outlook)
|
||||
print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}")
|
||||
print(f"{Colors.GRAY} Disclaimer: For educational/research purposes only.{Colors.RESET}")
|
||||
print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n")
|
||||
|
||||
@@ -1,439 +0,0 @@
|
||||
"""
|
||||
ForexBot2 — AI market outlook (advisory hourly assessment)
|
||||
===========================================================
|
||||
Builds a broad XAU/USD market brief and asks DeepSeek whether a good setup
|
||||
appears to be forming. The result is written to the signal-history database
|
||||
and may be displayed by consumers, but it is ADVISORY-ONLY: it never changes
|
||||
signal scoring, gates, grades, or trade setup generation.
|
||||
|
||||
The outlook runs outside the replayable signal path. Individual candle/news
|
||||
failures produce a partial brief; an unavailable LLM, malformed reply, or
|
||||
locked database never reaches a scan caller as an exception.
|
||||
|
||||
The static prompt is intentional: DeepSeek prices byte-identical prompt
|
||||
prefixes at its cache-hit rate, so timestamps and all other changing evidence
|
||||
live only in the JSON user payload.
|
||||
"""
|
||||
|
||||
import datetime as _dt
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from agent_panel import _normalize_candles
|
||||
from app_config import (get_app_config, market_outlook_enabled,
|
||||
signal_log_path)
|
||||
from config import PlatformConfig, Timeframe, XAUUSD
|
||||
from engine import TechnicalEngine
|
||||
from llm_client import get_llm_client
|
||||
from providers import base as _provider_base
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_MAX_PER_DAY = 12
|
||||
DEFAULT_CANDLE_BARS = 20
|
||||
_OUTLOOK_TFS = (Timeframe.M15, Timeframe.H1, Timeframe.H4)
|
||||
_VALID_BIASES = ("bullish", "bearish", "neutral")
|
||||
_VALID_SETUPS = ("buy", "sell", "none")
|
||||
|
||||
|
||||
# STATIC prompt: repeated prompt prefixes are much cheaper with DeepSeek's
|
||||
# server-side cache. Variable market evidence belongs exclusively in payload.
|
||||
SYSTEM_PROMPT = """\
|
||||
You are writing an advisory market outlook for gold (XAU/USD). You receive a
|
||||
JSON market brief covering recent candle shape, indicator states, price
|
||||
levels, news, and recent technical signals. Assess what may be forming soon;
|
||||
you are not a signal generator and must not provide an entry price, stop, or
|
||||
target.
|
||||
|
||||
Gold doctrine: hawkish Fed / rate hikes / rising real yields / strong
|
||||
dollar are gold-bearish; easing / rate cuts / safe-haven flows /
|
||||
geopolitical stress / weak dollar are gold-bullish.
|
||||
|
||||
Indicator "value" units: rsi 0-100 (above 70 overbought, below 30
|
||||
oversold); adx 0-100 (above 25 = strong trend); macd = histogram in
|
||||
dollars (momentum); roc = percent; bollinger = position 0-1 (0 = lower
|
||||
band, 1 = upper band); keltner = dollar distance from the mid-line;
|
||||
ema_trend = 0-1 alignment score; divergence and candlestick = pattern
|
||||
counts. "candles" is per timeframe: {"ref_close", "atr", "bars"}; bars
|
||||
are oldest-first [open, high, low, close] in ATR units relative to the
|
||||
latest close. Read their shape, never quote them as prices.
|
||||
|
||||
Hard rules: use ONLY the JSON provided. Do not invent numbers, events, or
|
||||
price levels. Your training data about gold, markets, and current events is
|
||||
stale and OFF-LIMITS: never use outside knowledge; if a fact is not in the
|
||||
JSON, treat it as unknown. If evidence is thin or missing, answer "neutral"
|
||||
with confidence below 30. Use plain everyday English for someone unfamiliar
|
||||
with markets: short simple sentences, no acronyms or jargon. Keep "watch"
|
||||
and "note" to at most 25 words each.
|
||||
|
||||
Reply with ONLY JSON:
|
||||
{"bias": "bullish"|"bearish"|"neutral",
|
||||
"setup_forming": "buy"|"sell"|"none",
|
||||
"confidence": 0-100,
|
||||
"watch": "<what to watch for, max 25 words>",
|
||||
"note": "<plain-English summary, max 25 words>"}
|
||||
"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class OutlookResult:
|
||||
bias: str
|
||||
setup_forming: str
|
||||
confidence: float
|
||||
watch: str
|
||||
note: str
|
||||
created_ts: float
|
||||
|
||||
|
||||
def _finite(x: Any) -> Optional[float]:
|
||||
"""float(x) or None for non-numeric / non-finite JSON values."""
|
||||
try:
|
||||
value = float(x)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return value if math.isfinite(value) else None
|
||||
|
||||
|
||||
def _round_finite(x: Any, digits: int) -> Optional[float]:
|
||||
value = _finite(x)
|
||||
return round(value, digits) if value is not None else None
|
||||
|
||||
|
||||
def _json_safe(value: Any) -> Any:
|
||||
"""Replace any accidental NaN/inf nested in provider data with null."""
|
||||
if isinstance(value, dict):
|
||||
return {str(key): _json_safe(item) for key, item in value.items()}
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [_json_safe(item) for item in value]
|
||||
if isinstance(value, bool) or value is None or isinstance(value, str):
|
||||
return value
|
||||
if isinstance(value, (int, float)):
|
||||
return _finite(value)
|
||||
return value
|
||||
|
||||
|
||||
def _clamp_confidence(raw: Any) -> Optional[float]:
|
||||
value = _finite(raw)
|
||||
return max(0.0, min(100.0, value)) if value is not None else None
|
||||
|
||||
|
||||
def _word_limit(text: Any, words: int = 25) -> str:
|
||||
"""Bound a response field by words as well as a practical char limit."""
|
||||
clean = str(text or "").strip()
|
||||
return " ".join(clean.split()[:words])[:200]
|
||||
|
||||
|
||||
def _validate_outlook(reply: Any) -> Optional[OutlookResult]:
|
||||
"""Validate only model replies that fully match the requested contract."""
|
||||
if not isinstance(reply, dict):
|
||||
return None
|
||||
bias = reply.get("bias")
|
||||
setup_forming = reply.get("setup_forming")
|
||||
confidence = _clamp_confidence(reply.get("confidence"))
|
||||
if (bias not in _VALID_BIASES or setup_forming not in _VALID_SETUPS
|
||||
or confidence is None):
|
||||
return None
|
||||
return OutlookResult(
|
||||
bias=bias,
|
||||
setup_forming=setup_forming,
|
||||
confidence=confidence,
|
||||
watch=_word_limit(reply.get("watch")),
|
||||
note=_word_limit(reply.get("note")),
|
||||
created_ts=time.time(),
|
||||
)
|
||||
|
||||
|
||||
# ── Data collection ──────────────────────────────────────────────────
|
||||
|
||||
_dp: Any = None # Patchable singleton, mirroring agent_panel._dp.
|
||||
|
||||
|
||||
def _outlook_provider() -> Any:
|
||||
"""Lazy provider singleton so its closed-bar cache stays warm hourly."""
|
||||
global _dp
|
||||
if _dp is None:
|
||||
from providers import make_data_provider
|
||||
_dp = make_data_provider(PlatformConfig())
|
||||
return _dp
|
||||
|
||||
|
||||
def _indicator_payload(analysis: Any) -> Dict[str, Dict[str, Any]]:
|
||||
"""Visible scored indicators with finite, compact values only."""
|
||||
out: Dict[str, Dict[str, Any]] = {}
|
||||
for name, indicator in (getattr(analysis, "indicators", None) or {}).items():
|
||||
if name.startswith("_") or getattr(indicator, "weight", 0) == 0:
|
||||
continue
|
||||
entry: Dict[str, Any] = {
|
||||
"signal": getattr(getattr(indicator, "signal", None), "name", ""),
|
||||
"confidence": _round_finite(
|
||||
getattr(indicator, "confidence", None), 1),
|
||||
"value": _round_finite(getattr(indicator, "value", None), 2),
|
||||
}
|
||||
out[name] = entry
|
||||
return out
|
||||
|
||||
|
||||
def _analysis_payload(analyses: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Pure formatter for the technical portion of the market brief."""
|
||||
indicators: Dict[str, Any] = {}
|
||||
market_regime: Dict[str, Optional[str]] = {}
|
||||
volatility_regime: Dict[str, Optional[str]] = {}
|
||||
key_levels: Dict[str, Dict[str, Optional[float]]] = {}
|
||||
for tf in _OUTLOOK_TFS:
|
||||
name = tf.value
|
||||
analysis = analyses.get(name)
|
||||
indicators[name] = _indicator_payload(analysis)
|
||||
market_regime[name] = getattr(analysis, "market_regime", None)
|
||||
volatility_regime[name] = getattr(analysis, "volatility_regime", None)
|
||||
key_levels[name] = {
|
||||
level: _round_finite(value, 2)
|
||||
for level, value in (getattr(analysis, "key_levels", None) or {}).items()
|
||||
}
|
||||
return {
|
||||
"indicators": indicators,
|
||||
"market_regime": market_regime,
|
||||
"volatility_regime": volatility_regime,
|
||||
"key_levels": key_levels,
|
||||
}
|
||||
|
||||
|
||||
def _empty_news_payload() -> Dict[str, Any]:
|
||||
return {
|
||||
"headlines": None,
|
||||
"events": None,
|
||||
"gold_sentiment": None,
|
||||
"remaining_high_impact_events": None,
|
||||
}
|
||||
|
||||
|
||||
def _news_payload() -> Dict[str, Any]:
|
||||
"""Collect current news without allowing any news failure to block us."""
|
||||
try:
|
||||
from news_engine import NewsEngine
|
||||
|
||||
news = NewsEngine(PlatformConfig().news)
|
||||
context = news.get_fundamental_context()
|
||||
headlines = [str(getattr(h, "headline", ""))
|
||||
for h in (context.news_headlines or [])[:10]
|
||||
if getattr(h, "headline", None)]
|
||||
events = [
|
||||
{
|
||||
"event": getattr(event, "event", None),
|
||||
"currency": getattr(event, "currency", None),
|
||||
"actual": getattr(event, "actual", None),
|
||||
"forecast": getattr(event, "forecast", None),
|
||||
"released": not event.is_upcoming,
|
||||
}
|
||||
for event in (context.events_today or [])
|
||||
if event.is_high_impact
|
||||
][:8]
|
||||
gold_sentiment, _ = news.get_pair_sentiment(XAUUSD, context)
|
||||
remaining = sum(
|
||||
1 for event in news.get_week_events()
|
||||
if event.is_high_impact and event.is_upcoming
|
||||
)
|
||||
return {
|
||||
"headlines": headlines,
|
||||
"events": events,
|
||||
"gold_sentiment": _round_finite(gold_sentiment, 2),
|
||||
"remaining_high_impact_events": remaining,
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.debug("market outlook: news context unavailable: %s", exc)
|
||||
return _empty_news_payload()
|
||||
|
||||
|
||||
def _recent_signals() -> List[Dict[str, Any]]:
|
||||
"""Read the latest signal history through SQLite's immutable URI."""
|
||||
try:
|
||||
path = os.path.abspath(signal_log_path())
|
||||
# mode=ro (not immutable=1): immutable ignores the WAL, so the
|
||||
# newest rows written by the live loggers were invisible here.
|
||||
uri = f"file:{path}?mode=ro"
|
||||
con = sqlite3.connect(uri, uri=True)
|
||||
try:
|
||||
rows = con.execute(
|
||||
"SELECT logged_at, timeframe, signal, quality, panel_verdict "
|
||||
"FROM signals ORDER BY id DESC LIMIT 10").fetchall()
|
||||
finally:
|
||||
con.close()
|
||||
return [
|
||||
{
|
||||
"logged_at": row[0],
|
||||
"timeframe": row[1],
|
||||
"signal": row[2],
|
||||
"quality": row[3],
|
||||
"panel_verdict": row[4],
|
||||
}
|
||||
for row in reversed(rows)
|
||||
]
|
||||
except Exception as exc:
|
||||
logger.debug("market outlook: signal history unavailable: %s", exc)
|
||||
return []
|
||||
|
||||
|
||||
def _build_payload(
|
||||
analyses: Dict[str, Any],
|
||||
candles: Dict[str, Any],
|
||||
news: Dict[str, Any],
|
||||
recent_signals: List[Dict[str, Any]],
|
||||
as_of: _dt.datetime,
|
||||
) -> Dict[str, Any]:
|
||||
"""Pure market-brief formatter. Every numeric field is finite or null."""
|
||||
technical = _analysis_payload(analyses)
|
||||
return _json_safe({
|
||||
"as_of": as_of.astimezone(_dt.timezone.utc).isoformat(),
|
||||
"candles": candles,
|
||||
**technical,
|
||||
"news": news,
|
||||
"recent_signals": recent_signals,
|
||||
})
|
||||
|
||||
|
||||
def _collect_payload(settings: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Fetch closed bars and turn them into a structured market brief."""
|
||||
analyses: Dict[str, Any] = {}
|
||||
candles: Dict[str, Any] = {tf.value: None for tf in _OUTLOOK_TFS}
|
||||
bars = int(settings.get("candles_bars", DEFAULT_CANDLE_BARS))
|
||||
provider = _outlook_provider()
|
||||
engine = TechnicalEngine(PlatformConfig().technical)
|
||||
|
||||
for tf in _OUTLOOK_TFS:
|
||||
try:
|
||||
df = provider.fetch_ohlcv(XAUUSD, tf)
|
||||
analysis = engine.analyze(df, XAUUSD, tf.value)
|
||||
analyses[tf.value] = analysis
|
||||
atr = _finite(getattr(
|
||||
(getattr(analysis, "indicators", None) or {}).get("atr"),
|
||||
"value", None))
|
||||
if atr is not None and atr > 0 and df is not None and len(df) >= 2:
|
||||
candles[tf.value] = _normalize_candles(df, atr, bars)
|
||||
except Exception as exc:
|
||||
logger.debug("market outlook: %s data unavailable: %s", tf.value, exc)
|
||||
|
||||
return _build_payload(
|
||||
analyses=analyses,
|
||||
candles=candles,
|
||||
news=_news_payload(),
|
||||
recent_signals=_recent_signals(),
|
||||
as_of=_dt.datetime.now(_dt.timezone.utc),
|
||||
)
|
||||
|
||||
|
||||
# ── Orchestration, hourly dedup, and request budget ──────────────────
|
||||
|
||||
_state_lock = threading.Lock()
|
||||
_last_run_hour: Optional[int] = None # UTC hour bucket of the last attempt
|
||||
_outlook_day: Optional[_dt.date] = None
|
||||
_outlook_count = 0
|
||||
_warned_budget = False
|
||||
|
||||
|
||||
def _skip(reason: str) -> None:
|
||||
"""Heartbeat for every silent early return — a dark outlook feature
|
||||
(zero rows for 15 days, 2026-08/09) must be visible in the log."""
|
||||
logger.info("market outlook: skipped (%s)", reason)
|
||||
|
||||
|
||||
def reset_outlook_state() -> None:
|
||||
"""Drop hourly and daily state (tests / deployment config reloads)."""
|
||||
global _last_run_hour, _outlook_day, _outlook_count
|
||||
global _warned_budget
|
||||
with _state_lock:
|
||||
_last_run_hour = None
|
||||
_outlook_day = None
|
||||
_outlook_count = 0
|
||||
_warned_budget = False
|
||||
|
||||
|
||||
def _begin_attempt(cap: int, force: bool) -> bool:
|
||||
"""Atomically apply hourly dedup, then reserve a daily attempt.
|
||||
|
||||
Hourly dedup is by UTC hour BUCKET, not ``now - last < 3600``: the
|
||||
H1 loop calls this at boundary+10 s+scan time, so consecutive hours
|
||||
were routinely 3,590 s apart and every other hour was dropped."""
|
||||
global _last_run_hour, _outlook_day, _outlook_count
|
||||
global _warned_budget
|
||||
now = time.time()
|
||||
hour_bucket = int(now // 3600)
|
||||
today = _dt.datetime.now(_dt.timezone.utc).date()
|
||||
with _state_lock:
|
||||
if not force and _last_run_hour == hour_bucket:
|
||||
_skip(f"hour bucket {hour_bucket} already attempted")
|
||||
return False
|
||||
if _outlook_day != today:
|
||||
_outlook_day = today
|
||||
_outlook_count = 0
|
||||
_warned_budget = False
|
||||
if _outlook_count >= cap:
|
||||
if not _warned_budget:
|
||||
_warned_budget = True
|
||||
logger.warning(
|
||||
"market outlook: daily budget (%d) exhausted — no more "
|
||||
"outlooks until UTC midnight", cap)
|
||||
else:
|
||||
_skip("daily budget exhausted")
|
||||
return False
|
||||
_outlook_count += 1
|
||||
_last_run_hour = hour_bucket
|
||||
return True
|
||||
|
||||
|
||||
def run_outlook(settings: Dict[str, Any],
|
||||
surface: str = "live") -> Optional[OutlookResult]:
|
||||
"""Build, ask, validate, and persist one advisory outlook."""
|
||||
try:
|
||||
payload = _collect_payload(settings)
|
||||
reply = get_llm_client().chat_json(SYSTEM_PROMPT, payload)
|
||||
result = _validate_outlook(reply)
|
||||
if result is None:
|
||||
logger.warning(
|
||||
"market outlook: no outlook produced (%s)",
|
||||
"LLM returned nothing — cap/timeout/disabled" if reply is None
|
||||
else "reply failed validation")
|
||||
return None
|
||||
try:
|
||||
from signal_log import log_outlook
|
||||
log_outlook(result, surface=surface, payload=payload)
|
||||
except Exception:
|
||||
logger.debug("market outlook: persistence unavailable", exc_info=True)
|
||||
return result
|
||||
except Exception:
|
||||
logger.warning("market outlook: run failed", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
def maybe_run_outlook(surface: str, force: bool = False) -> Optional[OutlookResult]:
|
||||
"""Best-effort public entry point. It NEVER raises to a scan caller.
|
||||
|
||||
``force`` is for explicit CLI use: it bypasses enabled, surface, and
|
||||
hourly gates, but not the independent outlook budget or shared LLM cap.
|
||||
"""
|
||||
try:
|
||||
settings = get_app_config().market_outlook_settings
|
||||
if not force:
|
||||
if not market_outlook_enabled():
|
||||
_skip("disabled")
|
||||
return None
|
||||
if surface == "telegram" and not settings.get("telegram", False):
|
||||
_skip("telegram surface off")
|
||||
return None
|
||||
if _provider_base._is_forex_weekend():
|
||||
_skip("weekend")
|
||||
return None
|
||||
cap = int(settings.get("max_per_day", DEFAULT_MAX_PER_DAY))
|
||||
if cap < 1:
|
||||
_skip("max_per_day < 1")
|
||||
return None
|
||||
if not _begin_attempt(cap, force):
|
||||
return None
|
||||
return run_outlook(settings, surface=surface)
|
||||
except Exception:
|
||||
logger.warning("market outlook: unavailable", exc_info=True)
|
||||
return None
|
||||
+1
-1
@@ -438,7 +438,7 @@ class NewsEngine:
|
||||
# here rather than in the fetch so the class-level cache and
|
||||
# the feed-health counts still describe what the feeds sent;
|
||||
# every consumer (aggregate score, summary, dashboard, panel,
|
||||
# outlook, signal log) reads context.news_headlines.
|
||||
# signal log) reads context.news_headlines.
|
||||
kept = [h for h in headlines if not _is_neutral_headline(h)]
|
||||
if len(kept) != len(headlines):
|
||||
logger.debug(
|
||||
|
||||
@@ -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())
|
||||
@@ -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())
|
||||
@@ -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()
|
||||
@@ -184,19 +184,6 @@ CREATE TABLE IF NOT EXISTS scan_outcomes (
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_scan_outcomes_tf_bar
|
||||
ON scan_outcomes (timeframe, bar_ts);
|
||||
CREATE TABLE IF NOT EXISTS outlooks (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
logged_at TEXT NOT NULL,
|
||||
surface TEXT NOT NULL,
|
||||
bias TEXT,
|
||||
setup_forming TEXT,
|
||||
confidence REAL,
|
||||
watch TEXT,
|
||||
note TEXT,
|
||||
news_json TEXT,
|
||||
indicators_json TEXT,
|
||||
candles_json TEXT
|
||||
);
|
||||
"""
|
||||
|
||||
# Additive migration for columns introduced after the original schema.
|
||||
@@ -298,12 +285,6 @@ SCAN_LABEL_TIME_BUDGET = 2.0
|
||||
# often than a candle close. Explicit calls (batch script, tests) ignore it.
|
||||
SCAN_LABEL_MIN_INTERVAL = 60.0
|
||||
|
||||
_ADDITIVE_OUTLOOK_COLS = (
|
||||
("news_json", "TEXT"),
|
||||
("indicators_json", "TEXT"),
|
||||
("candles_json", "TEXT"),
|
||||
)
|
||||
|
||||
_SIGNAL_INSERT_COLS = (
|
||||
"logged_at", "source", "symbol", "timeframe", "signal", "score",
|
||||
"confidence", "quality", "action", "price", "spread", "market_regime",
|
||||
@@ -323,13 +304,6 @@ _INSERT = (
|
||||
+ ", ".join("?" for _ in _SIGNAL_INSERT_COLS) + ")"
|
||||
)
|
||||
|
||||
_OUTLOOK_INSERT = """
|
||||
INSERT INTO outlooks (
|
||||
logged_at, surface, bias, setup_forming, confidence, watch, note,
|
||||
news_json, indicators_json, candles_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
"""
|
||||
|
||||
|
||||
def _finite(x: Any) -> Optional[float]:
|
||||
"""float(x) or None for non-numeric / non-finite values (NaN/inf —
|
||||
@@ -851,7 +825,6 @@ class SignalLogger:
|
||||
self._with_lock_retry(
|
||||
lambda: self._conn.executescript(_SCHEMA))
|
||||
self._with_lock_retry(self._migrate_signal_columns)
|
||||
self._with_lock_retry(self._migrate_outlook_columns)
|
||||
self._with_lock_retry(self._ensure_news_view)
|
||||
|
||||
def _ensure_news_view(self) -> None:
|
||||
@@ -917,21 +890,6 @@ class SignalLogger:
|
||||
" ON signals (timeframe, bar_ts)")
|
||||
self._conn.commit()
|
||||
|
||||
def _migrate_outlook_columns(self) -> None:
|
||||
"""Add snapshot columns to pre-existing outlook tables."""
|
||||
existing = {
|
||||
r[1] for r in self._conn.execute("PRAGMA table_info(outlooks)")
|
||||
}
|
||||
for name, col_type in _ADDITIVE_OUTLOOK_COLS:
|
||||
if name not in existing:
|
||||
try:
|
||||
self._conn.execute(
|
||||
f"ALTER TABLE outlooks ADD COLUMN {name} {col_type}")
|
||||
except sqlite3.OperationalError as exc:
|
||||
if "duplicate column name" not in str(exc).lower():
|
||||
raise
|
||||
self._conn.commit()
|
||||
|
||||
def log(
|
||||
self,
|
||||
report: Any,
|
||||
@@ -1536,33 +1494,6 @@ class SignalLogger:
|
||||
self._conn.commit()
|
||||
return reseeded
|
||||
|
||||
def log_outlook(self, outlook: Any, surface: str,
|
||||
payload: Any = None) -> int:
|
||||
"""Insert one advisory market outlook; returns its rowid."""
|
||||
payload = payload if isinstance(payload, dict) else {}
|
||||
technical = {
|
||||
key: payload.get(key)
|
||||
for key in ("indicators", "market_regime",
|
||||
"volatility_regime", "key_levels")
|
||||
if key in payload
|
||||
}
|
||||
row = (
|
||||
datetime.now(timezone.utc).isoformat(),
|
||||
surface,
|
||||
getattr(outlook, "bias", None),
|
||||
getattr(outlook, "setup_forming", None),
|
||||
_finite(getattr(outlook, "confidence", None)),
|
||||
getattr(outlook, "watch", None),
|
||||
getattr(outlook, "note", None),
|
||||
_dumps(payload.get("news")) if "news" in payload else None,
|
||||
_dumps(technical) if technical else None,
|
||||
_dumps(payload.get("candles")) if "candles" in payload else None,
|
||||
)
|
||||
with self._lock:
|
||||
cur = self._conn.execute(_OUTLOOK_INSERT, row)
|
||||
self._conn.commit()
|
||||
return int(cur.lastrowid)
|
||||
|
||||
def close(self) -> None:
|
||||
with self._lock:
|
||||
self._conn.close()
|
||||
@@ -1700,28 +1631,6 @@ def log_watch_update(report: Any, watch: Any) -> None:
|
||||
exc_info=True)
|
||||
|
||||
|
||||
def log_outlook(outlook: Any, surface: str = "live",
|
||||
payload: Any = None) -> None:
|
||||
"""Persist one advisory outlook when signal logging is enabled.
|
||||
|
||||
Outlook persistence is deliberately independent of a signal row: the
|
||||
feature runs on its own hourly cadence and must never affect a scan.
|
||||
"""
|
||||
try:
|
||||
if not app_config.signal_log_enabled():
|
||||
return
|
||||
path = app_config.signal_log_path()
|
||||
with _loggers_lock:
|
||||
lg = _loggers.get(path)
|
||||
if lg is None:
|
||||
lg = SignalLogger(path)
|
||||
_loggers[path] = lg
|
||||
lg.log_outlook(outlook, surface, payload=payload)
|
||||
except Exception:
|
||||
logger.warning("signal_log: failed to persist market outlook",
|
||||
exc_info=True)
|
||||
|
||||
|
||||
def reset_loggers_for_tests() -> None:
|
||||
"""Close and drop all cached loggers (test isolation)."""
|
||||
with _loggers_lock:
|
||||
|
||||
@@ -75,7 +75,6 @@ from providers import BaseDataProvider, make_data_provider
|
||||
from providers.base import _is_daily_maintenance
|
||||
from agent_panel import attach_panel, panel_health
|
||||
from llm_client import get_llm_client
|
||||
from market_outlook import maybe_run_outlook
|
||||
from news_engine import FundamentalContext, NewsEngine
|
||||
from pair_signals import PAIR_PRODUCERS
|
||||
from signal_aggregator import SignalAggregator, SignalReport
|
||||
@@ -919,28 +918,6 @@ async def _scan_timeframe(
|
||||
_commit_alert(key, report)
|
||||
logger.info(_alert_sent_line(report, tf.value, delivered))
|
||||
|
||||
# This independent advisory call runs only after the H1 scan completes.
|
||||
# Offloading the LLM request keeps the event loop responsive.
|
||||
if tf == Timeframe.H1:
|
||||
try:
|
||||
outlook = await asyncio.get_running_loop().run_in_executor(
|
||||
_executor, maybe_run_outlook, "telegram")
|
||||
settings = get_app_config().market_outlook_settings
|
||||
if (outlook is not None and settings.get("telegram", False)
|
||||
and OWNER_ID is not None):
|
||||
forming = {
|
||||
"buy": "possible buy forming",
|
||||
"sell": "possible sell forming",
|
||||
"none": "no clear setup forming",
|
||||
}[outlook.setup_forming]
|
||||
message = (
|
||||
f"🔭 AI outlook: {outlook.bias}, {forming} "
|
||||
f"({outlook.confidence:.0f}%) — {_escape_md(outlook.note)}"
|
||||
)
|
||||
await _send_alert(bot, OWNER_ID, message)
|
||||
except Exception:
|
||||
logger.exception("Hourly AI market outlook failed")
|
||||
|
||||
|
||||
# ── WebSocket level-breach monitor ──────────────────────────────────
|
||||
|
||||
|
||||
+86
-497
@@ -3323,6 +3323,36 @@ def _():
|
||||
f"example config.json must be byte-identical to defaults on {tf}"
|
||||
|
||||
|
||||
@test("app_config: a legacy market_outlook block is tolerated and ignored")
|
||||
def _():
|
||||
# The market-outlook feature was removed 2026-09-15; config.json files
|
||||
# on the live box still carry its block and must keep loading (the
|
||||
# oanda.account_id precedent), with one warning so the owner deletes it.
|
||||
import logging
|
||||
records = []
|
||||
|
||||
class _Grab(logging.Handler):
|
||||
def emit(self, record):
|
||||
records.append(record)
|
||||
|
||||
log = logging.getLogger("app_config")
|
||||
handler = _Grab()
|
||||
log.addHandler(handler)
|
||||
path = _write_json({
|
||||
"market_outlook": {"enabled": True, "telegram": True,
|
||||
"max_per_day": 12, "candles_bars": 20},
|
||||
})
|
||||
try:
|
||||
app = load_app_config(path)
|
||||
finally:
|
||||
log.removeHandler(handler)
|
||||
os.remove(path)
|
||||
assert not hasattr(app, "market_outlook_settings")
|
||||
assert any(r.levelno == logging.WARNING
|
||||
and "market_outlook" in r.getMessage() for r in records), \
|
||||
"loading a legacy market_outlook block must log one warning"
|
||||
|
||||
|
||||
@test("providers: api key resolves env → config.json (no code default)")
|
||||
def _():
|
||||
from providers import TwelveDataProvider
|
||||
@@ -4514,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")
|
||||
@@ -4573,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)
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
@@ -8528,409 +8522,6 @@ def _():
|
||||
os.remove(path + suffix)
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# MODULE 26: market_outlook.py (advisory hourly market brief)
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
||||
print("\n═══ Module 26: AI market outlook ═══")
|
||||
|
||||
import market_outlook as mo
|
||||
|
||||
|
||||
class _OutlookStub:
|
||||
available = True
|
||||
|
||||
def __init__(self, reply=None):
|
||||
self.reply = reply or {
|
||||
"bias": "bullish", "setup_forming": "buy", "confidence": 68,
|
||||
"watch": "Look for the price to hold above its recent low.",
|
||||
"note": "The short-term chart is improving while the bigger trend remains supportive.",
|
||||
}
|
||||
self.calls = []
|
||||
|
||||
def chat_json(self, system_prompt, user_payload):
|
||||
self.calls.append((system_prompt, user_payload))
|
||||
if isinstance(self.reply, Exception):
|
||||
raise self.reply
|
||||
return self.reply
|
||||
|
||||
|
||||
class _OutlookNews:
|
||||
def __init__(self, config):
|
||||
pass
|
||||
|
||||
def get_fundamental_context(self):
|
||||
now = datetime.now(timezone.utc)
|
||||
return FundamentalContext(
|
||||
news_headlines=[NewsSentiment(
|
||||
headline="Gold firms as the dollar weakens", source="test",
|
||||
timestamp=now, currencies_mentioned=["XAU", "USD"],
|
||||
sentiment_score=0.3, keywords=[])],
|
||||
events_today=[EconomicEvent(
|
||||
time=now - timedelta(minutes=5), currency="USD", impact="high",
|
||||
event="Inflation", actual="0.2%", forecast="0.3%")],
|
||||
)
|
||||
|
||||
def get_pair_sentiment(self, symbol, context):
|
||||
return 0.25, "slightly bullish"
|
||||
|
||||
def get_week_events(self):
|
||||
return []
|
||||
|
||||
|
||||
_OUTLOOK_ON = {
|
||||
"enabled": True,
|
||||
"telegram": True,
|
||||
"max_per_day": 12,
|
||||
"candles_bars": 20,
|
||||
}
|
||||
|
||||
|
||||
@test("app_config: market_outlook defaults, env override, and validation")
|
||||
def _():
|
||||
defaults = load_app_config(_write_json({})).market_outlook_settings
|
||||
assert defaults == {
|
||||
"enabled": False, "telegram": False,
|
||||
"max_per_day": 12, "candles_bars": 20,
|
||||
}
|
||||
app = load_app_config(_write_json({"market_outlook": _OUTLOOK_ON}))
|
||||
assert app.market_outlook_settings == _OUTLOOK_ON
|
||||
old = os.environ.get("MARKET_OUTLOOK_ENABLED")
|
||||
try:
|
||||
os.environ["MARKET_OUTLOOK_ENABLED"] = "false"
|
||||
assert app_config.market_outlook_enabled(app) is False
|
||||
os.environ["MARKET_OUTLOOK_ENABLED"] = "true"
|
||||
assert app_config.market_outlook_enabled(app) is True
|
||||
finally:
|
||||
if old is None:
|
||||
os.environ.pop("MARKET_OUTLOOK_ENABLED", None)
|
||||
else:
|
||||
os.environ["MARKET_OUTLOOK_ENABLED"] = old
|
||||
for bad in (
|
||||
{"market_outlook": {"enabld": True}},
|
||||
{"market_outlook": {"enabled": "yes"}},
|
||||
{"market_outlook": {"telegram": 1}},
|
||||
{"market_outlook": {"max_per_day": 0}},
|
||||
{"market_outlook": {"max_per_day": True}},
|
||||
{"market_outlook": {"candles_bars": 2.5}},
|
||||
):
|
||||
try:
|
||||
load_app_config(_write_json(bad))
|
||||
assert False, f"should have rejected {bad}"
|
||||
except AppConfigError:
|
||||
pass
|
||||
|
||||
|
||||
@test("signal_log: market outlook table roundtrip and old-db migration")
|
||||
def _():
|
||||
import sqlite3
|
||||
import signal_log as slog
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
try:
|
||||
con = sqlite3.connect(path)
|
||||
# Legacy pre-panel schema: it has a signals table but no outlooks
|
||||
# table. Keep the index columns from the original schema present.
|
||||
con.execute(
|
||||
"CREATE TABLE signals (id INTEGER PRIMARY KEY, logged_at TEXT,"
|
||||
" symbol TEXT, timeframe TEXT)")
|
||||
con.execute(
|
||||
"CREATE TABLE outlooks (id INTEGER PRIMARY KEY, logged_at TEXT,"
|
||||
" surface TEXT, bias TEXT, setup_forming TEXT, confidence REAL,"
|
||||
" watch TEXT, note TEXT)")
|
||||
con.commit()
|
||||
con.close()
|
||||
barrier = threading.Barrier(4)
|
||||
loggers = []
|
||||
errors = []
|
||||
|
||||
def open_logger():
|
||||
try:
|
||||
barrier.wait()
|
||||
loggers.append(slog.SignalLogger(path))
|
||||
except Exception as exc:
|
||||
errors.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=open_logger) for _ in range(4)]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
for thread in threads:
|
||||
thread.join(timeout=5)
|
||||
assert not errors, f"concurrent additive migration failed: {errors}"
|
||||
assert len(loggers) == 4
|
||||
logger = loggers[0]
|
||||
columns = {
|
||||
row[1] for row in logger._conn.execute("PRAGMA table_info(signals)")
|
||||
}
|
||||
assert {
|
||||
"indicators_json", "candles_json", "panel_created_at",
|
||||
"panel_cache_status", "panel_payload_hash",
|
||||
"advisory_outcome_status", "advisory_r_multiple",
|
||||
"advisory_outcome_json",
|
||||
} <= columns
|
||||
outlook_columns = {
|
||||
row[1] for row in logger._conn.execute("PRAGMA table_info(outlooks)")
|
||||
}
|
||||
assert {"news_json", "indicators_json", "candles_json"} <= \
|
||||
outlook_columns
|
||||
result = mo.OutlookResult(
|
||||
bias="bearish", setup_forming="sell", confidence=71.5,
|
||||
watch="Wait for price to fail at resistance.",
|
||||
note="Momentum is fading after a sharp rise.", created_ts=time.time())
|
||||
payload = {
|
||||
"news": {"headlines": ["Gold firms"]},
|
||||
"indicators": {"1h": {"rsi": {"value": 55.0}}},
|
||||
"market_regime": {"1h": "trending_up"},
|
||||
"volatility_regime": {"1h": "normal"},
|
||||
"key_levels": {"1h": {"support": 4400.0}},
|
||||
"candles": {"1h": {"bars": [[-0.2, 0.1, -0.3, 0.0]]}},
|
||||
}
|
||||
rowid = logger.log_outlook(result, surface="test", payload=payload)
|
||||
for open_instance in loggers:
|
||||
open_instance.close()
|
||||
con = sqlite3.connect(path)
|
||||
row = con.execute(
|
||||
"SELECT surface, bias, setup_forming, confidence, watch, note, "
|
||||
"news_json, indicators_json, candles_json "
|
||||
"FROM outlooks WHERE id=?", (rowid,)).fetchone()
|
||||
con.close()
|
||||
assert row[:6] == ("test", "bearish", "sell", 71.5,
|
||||
"Wait for price to fail at resistance.",
|
||||
"Momentum is fading after a sharp rise.")
|
||||
assert json.loads(row[6])["headlines"] == ["Gold firms"]
|
||||
assert json.loads(row[7])["market_regime"]["1h"] == "trending_up"
|
||||
assert len(json.loads(row[8])["1h"]["bars"]) == 1
|
||||
finally:
|
||||
slog.reset_loggers_for_tests()
|
||||
os.remove(path)
|
||||
|
||||
|
||||
@test("market_outlook: payload is finite, normalized, and news-safe")
|
||||
def _():
|
||||
try:
|
||||
mo._dp = _FakeDP()
|
||||
with patch("news_engine.NewsEngine", _OutlookNews):
|
||||
payload = mo._collect_payload(_OUTLOOK_ON)
|
||||
assert mo._normalize_candles is ap._normalize_candles
|
||||
assert set(payload["candles"]) == {"15min", "1h", "4h"}
|
||||
assert all(len(block["bars"]) == 20
|
||||
for block in payload["candles"].values())
|
||||
assert payload["indicators"]["1h"]
|
||||
assert payload["news"]["headlines"] == ["Gold firms as the dollar weakens"]
|
||||
json.dumps(payload, sort_keys=True, allow_nan=False)
|
||||
|
||||
with patch("news_engine.NewsEngine", side_effect=RuntimeError("down")):
|
||||
no_news = mo._collect_payload(_OUTLOOK_ON)["news"]
|
||||
assert no_news == mo._empty_news_payload()
|
||||
finally:
|
||||
mo._dp = None
|
||||
|
||||
|
||||
@test("market_outlook: reply validation clamps and rejects malformed JSON")
|
||||
def _():
|
||||
validate = mo._validate_outlook
|
||||
assert validate(None) is None
|
||||
assert validate({"bias": "up", "setup_forming": "buy", "confidence": 50}) is None
|
||||
assert validate({"bias": "bullish", "setup_forming": "wait", "confidence": 50}) is None
|
||||
assert validate({"bias": "bullish", "setup_forming": "buy"}) is None
|
||||
assert validate({"bias": "bullish", "setup_forming": "buy", "confidence": "hot"}) is None
|
||||
result = validate({
|
||||
"bias": "neutral", "setup_forming": "none", "confidence": 150,
|
||||
"watch": "word " * 30, "note": "note " * 30,
|
||||
})
|
||||
assert result.confidence == 100.0
|
||||
assert len(result.watch.split()) == 25 and len(result.note.split()) == 25
|
||||
|
||||
|
||||
@test("market_outlook: disabled default spends no LLM request")
|
||||
def _():
|
||||
original_app = app_config.get_app_config()
|
||||
old = os.environ.pop("MARKET_OUTLOOK_ENABLED", None)
|
||||
try:
|
||||
app_config.set_app_config(AppConfig())
|
||||
stub = _OutlookStub()
|
||||
lc._client = stub
|
||||
mo.reset_outlook_state()
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
assert stub.calls == []
|
||||
finally:
|
||||
if old is not None:
|
||||
os.environ["MARKET_OUTLOOK_ENABLED"] = old
|
||||
lc._client = None
|
||||
mo.reset_outlook_state()
|
||||
app_config.set_app_config(original_app)
|
||||
|
||||
|
||||
@test("market_outlook: hourly dedup prevents a second request")
|
||||
def _():
|
||||
original_app = app_config.get_app_config()
|
||||
old = os.environ.pop("MARKET_OUTLOOK_ENABLED", None)
|
||||
try:
|
||||
app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON)))
|
||||
stub = _OutlookStub()
|
||||
lc._client = stub
|
||||
mo.reset_outlook_state()
|
||||
with patch("providers.base._is_forex_weekend", return_value=False), \
|
||||
patch.object(mo, "_collect_payload", return_value={}), \
|
||||
patch("signal_log.log_outlook"):
|
||||
assert mo.maybe_run_outlook("telegram") is not None
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
assert len(stub.calls) == 1
|
||||
finally:
|
||||
if old is not None:
|
||||
os.environ["MARKET_OUTLOOK_ENABLED"] = old
|
||||
lc._client = None
|
||||
mo.reset_outlook_state()
|
||||
app_config.set_app_config(original_app)
|
||||
|
||||
|
||||
@test("market_outlook: daily budget blocks force calls after the cap")
|
||||
def _():
|
||||
original_app = app_config.get_app_config()
|
||||
try:
|
||||
app_config.set_app_config(AppConfig(market_outlook_settings={
|
||||
**_OUTLOOK_ON, "max_per_day": 1}))
|
||||
stub = _OutlookStub()
|
||||
lc._client = stub
|
||||
mo.reset_outlook_state()
|
||||
with patch("providers.base._is_forex_weekend", return_value=False), \
|
||||
patch.object(mo, "_collect_payload", return_value={}), \
|
||||
patch("signal_log.log_outlook"):
|
||||
assert mo.maybe_run_outlook("cli", force=True) is not None
|
||||
assert mo.maybe_run_outlook("cli", force=True) is None
|
||||
assert len(stub.calls) == 1
|
||||
finally:
|
||||
lc._client = None
|
||||
mo.reset_outlook_state()
|
||||
app_config.set_app_config(original_app)
|
||||
|
||||
|
||||
@test("market_outlook: LLM and persistence failures remain fail-safe")
|
||||
def _():
|
||||
original_app = app_config.get_app_config()
|
||||
try:
|
||||
app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON)))
|
||||
lc._client = _OutlookStub(RuntimeError("LLM down"))
|
||||
mo.reset_outlook_state()
|
||||
with patch("providers.base._is_forex_weekend", return_value=False), \
|
||||
patch.object(mo, "_collect_payload", return_value={}):
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
|
||||
lc._client = _OutlookStub()
|
||||
mo.reset_outlook_state()
|
||||
with patch("providers.base._is_forex_weekend", return_value=False), \
|
||||
patch.object(mo, "_collect_payload", return_value={}), \
|
||||
patch("signal_log.log_outlook", side_effect=RuntimeError("db locked")):
|
||||
assert mo.maybe_run_outlook("telegram") is not None
|
||||
finally:
|
||||
lc._client = None
|
||||
mo.reset_outlook_state()
|
||||
app_config.set_app_config(original_app)
|
||||
|
||||
|
||||
@test("market_outlook: weekend gate spends nothing")
|
||||
def _():
|
||||
original_app = app_config.get_app_config()
|
||||
try:
|
||||
app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON)))
|
||||
stub = _OutlookStub()
|
||||
lc._client = stub
|
||||
mo.reset_outlook_state()
|
||||
with patch("providers.base._is_forex_weekend", return_value=True):
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
assert stub.calls == []
|
||||
finally:
|
||||
lc._client = None
|
||||
mo.reset_outlook_state()
|
||||
app_config.set_app_config(original_app)
|
||||
|
||||
|
||||
@test("market_outlook: hour-bucket dedup runs consecutive hours 3,590 s apart")
|
||||
def _():
|
||||
original_app = app_config.get_app_config()
|
||||
old = os.environ.pop("MARKET_OUTLOOK_ENABLED", None)
|
||||
try:
|
||||
app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON)))
|
||||
stub = _OutlookStub()
|
||||
lc._client = stub
|
||||
mo.reset_outlook_state()
|
||||
base = 1_800_000_000 - (1_800_000_000 % 3600) # an exact hour
|
||||
clock = {"t": base + 25.0} # 10:00:25
|
||||
with patch("providers.base._is_forex_weekend", return_value=False), \
|
||||
patch.object(mo, "_collect_payload", return_value={}), \
|
||||
patch("signal_log.log_outlook"), \
|
||||
patch.object(mo.time, "time", lambda: clock["t"]):
|
||||
assert mo.maybe_run_outlook("telegram") is not None
|
||||
clock["t"] = base + 1800.0 # 10:30 same bucket
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
clock["t"] = base + 3600.0 + 15.0 # 11:00:15 — 3,590 s later
|
||||
assert mo.maybe_run_outlook("telegram") is not None, \
|
||||
"the old `now - last < 3600` rule dropped every other hour"
|
||||
assert len(stub.calls) == 2
|
||||
finally:
|
||||
if old is not None:
|
||||
os.environ["MARKET_OUTLOOK_ENABLED"] = old
|
||||
lc._client = None
|
||||
mo.reset_outlook_state()
|
||||
app_config.set_app_config(original_app)
|
||||
|
||||
|
||||
@test("market_outlook: every silent early return leaves a heartbeat log line")
|
||||
def _():
|
||||
original_app = app_config.get_app_config()
|
||||
old = os.environ.pop("MARKET_OUTLOOK_ENABLED", None)
|
||||
records = []
|
||||
|
||||
class _Grab(logging.Handler):
|
||||
def emit(self, record):
|
||||
records.append(record.getMessage())
|
||||
|
||||
handler = _Grab()
|
||||
mo.logger.addHandler(handler)
|
||||
prev_level = mo.logger.level
|
||||
mo.logger.setLevel(logging.INFO)
|
||||
try:
|
||||
app_config.set_app_config(AppConfig(market_outlook_settings={
|
||||
**_OUTLOOK_ON, "enabled": False}))
|
||||
mo.reset_outlook_state()
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
assert any("skipped (disabled)" in m for m in records), records
|
||||
records.clear()
|
||||
app_config.set_app_config(AppConfig(market_outlook_settings=dict(_OUTLOOK_ON)))
|
||||
with patch("providers.base._is_forex_weekend", return_value=True):
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
assert any("skipped (weekend)" in m for m in records), records
|
||||
records.clear()
|
||||
# LLM returns nothing → warning, nothing persisted
|
||||
stub = _OutlookStub(reply=None)
|
||||
stub.reply = None
|
||||
lc._client = stub
|
||||
persisted = []
|
||||
with patch("providers.base._is_forex_weekend", return_value=False), \
|
||||
patch.object(mo, "_collect_payload", return_value={}), \
|
||||
patch("signal_log.log_outlook", persisted.append):
|
||||
assert mo.maybe_run_outlook("telegram") is None
|
||||
assert any("no outlook produced" in m for m in records), records
|
||||
assert persisted == []
|
||||
finally:
|
||||
mo.logger.removeHandler(handler)
|
||||
mo.logger.setLevel(prev_level)
|
||||
if old is not None:
|
||||
os.environ["MARKET_OUTLOOK_ENABLED"] = old
|
||||
lc._client = None
|
||||
mo.reset_outlook_state()
|
||||
app_config.set_app_config(original_app)
|
||||
|
||||
|
||||
@test("market_outlook isolation: signal path never imports the advisory module")
|
||||
def _():
|
||||
root = os.path.dirname(__file__)
|
||||
for name in ("engine.py", "signal_aggregator.py"):
|
||||
with open(os.path.join(root, name), encoding="utf-8") as fh:
|
||||
assert "market_outlook" not in fh.read(), name
|
||||
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# MODULE 27: signal-log observability (2026-09 audit)
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
@@ -9200,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:
|
||||
@@ -9232,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()
|
||||
@@ -9500,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
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
@@ -10536,6 +10073,59 @@ def _():
|
||||
assert weighted_agg.gate_config_fingerprint() == m5.gate_config_fingerprint()
|
||||
|
||||
|
||||
@test("signal_log: concurrent legacy-DB migration adds the panel/advisory columns")
|
||||
def _():
|
||||
# Four loggers open the same pre-panel database at once: the additive
|
||||
# ALTER TABLE migration must survive the race (_with_lock_retry / WAL).
|
||||
# Moved from the removed market-outlook module (2026-09-15); it was the
|
||||
# suite's only concurrent-migration regression.
|
||||
import sqlite3
|
||||
import signal_log as slog
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
try:
|
||||
con = sqlite3.connect(path)
|
||||
# Legacy pre-panel schema. Keep the index columns from the original
|
||||
# schema present.
|
||||
con.execute(
|
||||
"CREATE TABLE signals (id INTEGER PRIMARY KEY, logged_at TEXT,"
|
||||
" symbol TEXT, timeframe TEXT)")
|
||||
con.commit()
|
||||
con.close()
|
||||
barrier = threading.Barrier(4)
|
||||
loggers = []
|
||||
errors = []
|
||||
|
||||
def open_logger():
|
||||
try:
|
||||
barrier.wait()
|
||||
loggers.append(slog.SignalLogger(path))
|
||||
except Exception as exc:
|
||||
errors.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=open_logger) for _ in range(4)]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
for thread in threads:
|
||||
thread.join(timeout=5)
|
||||
assert not errors, f"concurrent additive migration failed: {errors}"
|
||||
assert len(loggers) == 4
|
||||
columns = {
|
||||
row[1] for row in loggers[0]._conn.execute("PRAGMA table_info(signals)")
|
||||
}
|
||||
assert {
|
||||
"indicators_json", "candles_json", "panel_created_at",
|
||||
"panel_cache_status", "panel_payload_hash",
|
||||
"advisory_outcome_status", "advisory_r_multiple",
|
||||
"advisory_outcome_json",
|
||||
} <= columns
|
||||
for open_instance in loggers:
|
||||
open_instance.close()
|
||||
finally:
|
||||
slog.reset_loggers_for_tests()
|
||||
os.remove(path)
|
||||
|
||||
|
||||
@test("signal_log: migration adds the research columns, scan_outcomes and indexes")
|
||||
def _():
|
||||
import sqlite3
|
||||
@@ -11906,8 +11496,7 @@ def _():
|
||||
# the backtester.
|
||||
banned = {"signal_log", "news_engine", "providers", "telegram_bot",
|
||||
"news_archive", "agent_panel", "violent_watch",
|
||||
"market_outlook", "gpu_backtest", "sklearn", "scipy",
|
||||
"lightgbm", "torch"}
|
||||
"gpu_backtest", "sklearn", "scipy", "lightgbm", "torch"}
|
||||
path = Path("backtest/replay_record.py")
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"), str(path))
|
||||
seen, offenders = set(), []
|
||||
|
||||
+1
-1
@@ -7,7 +7,7 @@ hour may be wild. It is a WARNING LINE and a standalone heads-up
|
||||
message — nothing here ever changes a signal, grade, gate, trade setup
|
||||
or backtest result.
|
||||
|
||||
Hard independence rule (mirrors market_outlook.py / agent_panel.py):
|
||||
Hard independence rule (mirrors agent_panel.py):
|
||||
this module is NEVER imported by `signal_aggregator`, `engine`, or
|
||||
anything under `backtest/`. Surfaces (telegram_bot, main, the dashboard
|
||||
publisher) attach the result POST-HOC via `attach_watch`, exactly the
|
||||
|
||||
Reference in New Issue
Block a user