audit 2026-09-13: eleven delivery/measurement fixes; books byte-identical
Signal path
- A shipped setup always names a direction: grade C needs confidence
>= 45 (gate 6c grants B at >= 45) while BUY/SELL needs >= 50, so C/B
reports reached the alert gate as NEUTRAL with a full setup. The
backtester traded them (the replay never reads final_signal) while
Telegram rejected NEUTRAL first — 5/91 trades, +3.7R of +22.2R, in
the H1 12-month book. _attach_core_setup stamps the setup's direction
and repairs the pre-gate action text; no-setup reports stay NEUTRAL.
- analyze_pair fails the scan when ANY required frame is missing
("DATA UNAVAILABLE:", scan_kind='failed'); a lost confirmation frame
used to run with an unearned -15 MTF penalty.
Outcomes and labels
- advisory_outcome.resolve_locked_floor: a gap the database cannot fill
resolves at the locked trailing stop (GAP_LOCKED, result.floor) when
the stop had crossed the entry; the rebuild fires on both gap reasons
(rows 915/920: +1.14R / +0.84R floors; resolved sum -2.46R -> -0.48R).
- label_scan_outcomes: SCAN_LABEL_MAX_AGE_H = 72 h, stale rows never
anchor the shared bar stream, labeled_at bumped on re-examination,
_stream_since warns at its row cap (the Labor-Day pile would have
blinded scan_outcomes around 2026-09-14).
Providers and scheduling
- The daily settlement pause (Mon-Thu 17:00 New York) is a scheduled
closure for _has_latest_closed_bar (whole-span label walk-back) and
_is_data_fresh (closed seconds are not age); bot and dashboard skip
scans whose just-closed label sits in the pause (61 false stale rows
in three nights).
- fetch_range retries a failed page in place (three failures on one
window raise) and logs an ERROR for interior gaps of >3 unscheduled
bars.
- pd.Timestamp.utcnow() replaced in providers/base.py and engine.py;
requirements pin pandas<4, numpy<3.
- telegram_bot: boundary loops share one scan per bar with the breach
handler (_remember_scan keys on the analysed bar; breach defers 20 s
after an M15 boundary); the rotating file handler attaches in main()
only (the test suite used to write into logs/telegram.log).
- dashboard_publisher: catch_up_timeframes re-scans boundaries missed
while a cycle overran or the process was down.
Backtester and engine guards
- _replay_span returns the per-span GPU cache-miss count; run_backtest
sums, logs it (must be 0) and writes cache_fallbacks to the manifest.
- regime_weight_table raises on a partial regime_weights_override.
Verification: 519 tests; M5/M15 2-yr, M15 OOS, H1/H4 3-yr books
byte-identical before and after; live signals.db untouched (checked on
a copy). CLAUDE.md carries the dated entry and the blocked-hours union
semantics that every book was validated under.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0139xz1J6YccndA2be6yFnEo
This commit is contained in:
@@ -348,6 +348,39 @@ def advance_outcome(
|
|||||||
return state
|
return state
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_locked_floor(state: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
"""If an UNRESOLVABLE gap outcome had already trailed its stop past the
|
||||||
|
entry, finish it at that stop as a FLOOR (the true outcome is at least
|
||||||
|
this good) instead of discarding it. Returns the state (mutated)."""
|
||||||
|
if state.get("status") != "UNRESOLVABLE":
|
||||||
|
return state
|
||||||
|
reason = str(state.get("reason", ""))
|
||||||
|
# Both gap reasons ("lost continuity" / "contain an internal gap")
|
||||||
|
# share this prefix; an invalid seed or evaluator error never floors.
|
||||||
|
if not reason.startswith("stored candle snapshots"):
|
||||||
|
return state
|
||||||
|
progress = state.get("progress")
|
||||||
|
setup = state.get("setup")
|
||||||
|
if not isinstance(progress, dict) or not isinstance(setup, dict):
|
||||||
|
return state
|
||||||
|
entry = _finite(progress.get("entry_price"))
|
||||||
|
stop = _finite(progress.get("current_stop"))
|
||||||
|
risk = _finite(setup.get("stop_distance"))
|
||||||
|
exit_at = progress.get("last_processed_bar")
|
||||||
|
if (entry is None or stop is None or risk is None or risk <= 0
|
||||||
|
or not exit_at):
|
||||||
|
return state
|
||||||
|
buy = setup.get("direction") == "BUY"
|
||||||
|
locked = stop >= entry if buy else stop <= entry
|
||||||
|
if not locked:
|
||||||
|
return state
|
||||||
|
_finish(state, "GAP_LOCKED", str(exit_at), stop)
|
||||||
|
state["result"]["floor"] = True
|
||||||
|
state["reason"] = (
|
||||||
|
f"{reason} — resolved at the locked trailing stop (floor)")
|
||||||
|
return state
|
||||||
|
|
||||||
|
|
||||||
def canonical_r(state: Dict[str, Any]) -> Optional[float]:
|
def canonical_r(state: Dict[str, Any]) -> Optional[float]:
|
||||||
result = state.get("result")
|
result = state.get("result")
|
||||||
value = result.get("net_r_065") if isinstance(result, dict) else None
|
value = result.get("net_r_065") if isinstance(result, dict) else None
|
||||||
|
|||||||
@@ -1540,6 +1540,10 @@ def _replay_span(span: Tuple[int, int]):
|
|||||||
is applied by the caller in a sequential pass over the returned
|
is applied by the caller in a sequential pass over the returned
|
||||||
candidates, so parallel and sequential runs produce identical books.
|
candidates, so parallel and sequential runs produce identical books.
|
||||||
Must stay behaviorally identical to the original sequential loop.
|
Must stay behaviorally identical to the original sequential loop.
|
||||||
|
|
||||||
|
Returns (skipped, candidates, bar_rows, cache_fallbacks) — the last
|
||||||
|
element is this span's GPU-cache-miss count, which the caller sums so
|
||||||
|
a parity hazard inside a worker process cannot stay invisible.
|
||||||
"""
|
"""
|
||||||
ctx = _REPLAY_CTX
|
ctx = _REPLAY_CTX
|
||||||
primary_df = ctx["primary_df"]
|
primary_df = ctx["primary_df"]
|
||||||
@@ -1550,6 +1554,12 @@ def _replay_span(span: Tuple[int, int]):
|
|||||||
bt_cfg = ctx["bt_cfg"]
|
bt_cfg = ctx["bt_cfg"]
|
||||||
primary_dur = ctx["primary_dur"]
|
primary_dur = ctx["primary_dur"]
|
||||||
|
|
||||||
|
# One worker process handles several spans, so the engine's cache-miss
|
||||||
|
# counter must be per span, not cumulative across them.
|
||||||
|
engine = aggregator.tech_engine
|
||||||
|
if hasattr(engine, "fallback_count"):
|
||||||
|
engine.fallback_count = 0
|
||||||
|
|
||||||
skipped: Dict[str, int] = {}
|
skipped: Dict[str, int] = {}
|
||||||
candidates: List[Tuple[int, Trade]] = []
|
candidates: List[Tuple[int, Trade]] = []
|
||||||
is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
|
is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
|
||||||
@@ -1719,8 +1729,9 @@ def _replay_span(span: Tuple[int, int]):
|
|||||||
if tag == "XAU_PB" else 0.0),
|
if tag == "XAU_PB" else 0.0),
|
||||||
)))
|
)))
|
||||||
|
|
||||||
return skipped, candidates, (replay_record.finish_bar_rows(bar_buf)
|
return (skipped, candidates,
|
||||||
if record_bars else None)
|
(replay_record.finish_bar_rows(bar_buf) if record_bars else None),
|
||||||
|
int(getattr(engine, "fallback_count", 0)))
|
||||||
|
|
||||||
|
|
||||||
def _reanchor_entries_to_raw(
|
def _reanchor_entries_to_raw(
|
||||||
@@ -1927,14 +1938,28 @@ def run_backtest(
|
|||||||
skipped_reasons: Dict[str, int] = {}
|
skipped_reasons: Dict[str, int] = {}
|
||||||
candidates: List[Tuple[int, Trade]] = []
|
candidates: List[Tuple[int, Trade]] = []
|
||||||
bar_chunks: List = []
|
bar_chunks: List = []
|
||||||
for skips, cands, bars in results:
|
cache_fallbacks = 0
|
||||||
|
for skips, cands, bars, fallbacks in results:
|
||||||
for reason, count in skips.items():
|
for reason, count in skips.items():
|
||||||
skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count
|
skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count
|
||||||
candidates.extend(cands)
|
candidates.extend(cands)
|
||||||
if bars is not None:
|
if bars is not None:
|
||||||
bar_chunks.append(bars)
|
bar_chunks.append(bars)
|
||||||
|
cache_fallbacks += fallbacks
|
||||||
candidates.sort(key=lambda pc: pc[0])
|
candidates.sort(key=lambda pc: pc[0])
|
||||||
|
|
||||||
|
# The workers' counters die with their processes, so surface the run
|
||||||
|
# total here and leave it on the parent engine for callers to read.
|
||||||
|
aggregator.tech_engine.fallback_count = cache_fallbacks
|
||||||
|
if cache_fallbacks > 0:
|
||||||
|
logger.warning(
|
||||||
|
f"GPU cache fallbacks: {cache_fallbacks} indicator recomputes "
|
||||||
|
f"(one per _compute_* override per missed bar) fell back to the "
|
||||||
|
f"live engine's pandas math (different EWM seeding) — parity "
|
||||||
|
f"hazard; see GPUTechnicalEngine._note_fallback")
|
||||||
|
else:
|
||||||
|
logger.info("GPU cache fallbacks: 0")
|
||||||
|
|
||||||
# ── Phase 2: entry cooldown — the only sequential rule.
|
# ── Phase 2: entry cooldown — the only sequential rule.
|
||||||
accepted = _apply_cooldown(candidates, bt_cfg.cooldown_bars)
|
accepted = _apply_cooldown(candidates, bt_cfg.cooldown_bars)
|
||||||
trades: List[Trade] = [t for _, t in accepted]
|
trades: List[Trade] = [t for _, t in accepted]
|
||||||
@@ -1985,7 +2010,8 @@ def run_backtest(
|
|||||||
skipped=skipped_reasons, n_candidates=len(candidates),
|
skipped=skipped_reasons, n_candidates=len(candidates),
|
||||||
workers=workers,
|
workers=workers,
|
||||||
backend="CuPy GPU" if _GPU_AVAILABLE else "NumPy CPU fallback",
|
backend="CuPy GPU" if _GPU_AVAILABLE else "NumPy CPU fallback",
|
||||||
cache_files=cache_names, replay_seconds=replay_seconds),
|
cache_files=cache_names, replay_seconds=replay_seconds,
|
||||||
|
cache_fallbacks=cache_fallbacks),
|
||||||
sim_df=sim_df, bar_chunks=bar_chunks or None)
|
sim_df=sim_df, bar_chunks=bar_chunks or None)
|
||||||
logger.info(f"--record: wrote run to {bt_cfg.record_dir}")
|
logger.info(f"--record: wrote run to {bt_cfg.record_dir}")
|
||||||
|
|
||||||
|
|||||||
@@ -1055,6 +1055,9 @@ def write_run(record_dir, *, trades, book_df, bt_cfg, platform_cfg, aggregator,
|
|||||||
"max_hold_bars": bt_cfg.max_hold_bars,
|
"max_hold_bars": bt_cfg.max_hold_bars,
|
||||||
"fixed_rr_tp": bt_cfg.fixed_rr_tp,
|
"fixed_rr_tp": bt_cfg.fixed_rr_tp,
|
||||||
"workers": run_info.get("workers"),
|
"workers": run_info.get("workers"),
|
||||||
|
# >0 means some bar-analyses fell back to the live engine's pandas
|
||||||
|
# math instead of the pre-computed cache (a parity hazard).
|
||||||
|
"cache_fallbacks": run_info.get("cache_fallbacks"),
|
||||||
"engine_fingerprint": engine_fp,
|
"engine_fingerprint": engine_fp,
|
||||||
"gate_fingerprint": gate_fp,
|
"gate_fingerprint": gate_fp,
|
||||||
"technical_config": dataclasses.asdict(platform_cfg.technical),
|
"technical_config": dataclasses.asdict(platform_cfg.technical),
|
||||||
|
|||||||
+92
-6
@@ -26,6 +26,11 @@ whichever timeframes just closed. Due-classification uses the RAW wake
|
|||||||
time — flooring 12:00:10 to the 5-minute grid yields the 12:00 boundary
|
time — flooring 12:00:10 to the 5-minute grid yields the 12:00 boundary
|
||||||
that just closed. Do NOT subtract the buffer first: 12:00:10 − 15 s
|
that just closed. Do NOT subtract the buffer first: 12:00:10 − 15 s
|
||||||
floors to 11:55 and silently skips every H1/H4 scan (v1 bug, 2026-08-02).
|
floors to 11:55 and silently skips every H1/H4 scan (v1 bug, 2026-08-02).
|
||||||
|
A cycle can overrun its 5-minute slot (or the process can be down), so
|
||||||
|
each wake also catches up the timeframes whose latest closed boundary is
|
||||||
|
newer than their last scanned one (`catch_up_timeframes`) — without it the
|
||||||
|
page served an hours-old H1/H4 report behind a fresh countdown
|
||||||
|
(2026-09-13 audit).
|
||||||
|
|
||||||
Scans run sequentially M5→M15→H1→H4 through one shared data provider so
|
Scans run sequentially M5→M15→H1→H4 through one shared data provider so
|
||||||
later timeframes hit the warm boundary cache — that ordering IS the
|
later timeframes hit the warm boundary cache — that ordering IS the
|
||||||
@@ -46,10 +51,12 @@ import os
|
|||||||
import shutil
|
import shutil
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timedelta, timezone
|
||||||
from logging.handlers import TimedRotatingFileHandler
|
from logging.handlers import TimedRotatingFileHandler
|
||||||
from typing import Dict, List, Optional
|
from typing import Dict, List, Optional
|
||||||
|
|
||||||
|
import pandas as pd
|
||||||
|
|
||||||
from agent_panel import attach_panel
|
from agent_panel import attach_panel
|
||||||
from app_config import get_app_config, violent_watch_enabled
|
from app_config import get_app_config, violent_watch_enabled
|
||||||
from config import PlatformConfig, Timeframe, XAUUSD
|
from config import PlatformConfig, Timeframe, XAUUSD
|
||||||
@@ -230,10 +237,60 @@ def due_timeframes(wake_dt: datetime) -> List[Timeframe]:
|
|||||||
"""Timeframes whose candle just closed at the boundary this wake
|
"""Timeframes whose candle just closed at the boundary this wake
|
||||||
belongs to. Takes the RAW wake time (boundary + buffer): flooring to
|
belongs to. Takes the RAW wake time (boundary + buffer): flooring to
|
||||||
the 5-minute grid yields the boundary that just passed. Never
|
the 5-minute grid yields the boundary that just passed. Never
|
||||||
subtract the buffer first — see module docstring (v1 bug)."""
|
subtract the buffer first — see module docstring (v1 bug).
|
||||||
|
|
||||||
|
A timeframe whose just-closed bar would carry a label inside gold's
|
||||||
|
daily settlement pause is skipped: no such bar exists, so the scan
|
||||||
|
would only re-read the previous one (2026-09-13 audit)."""
|
||||||
minutes = wake_dt.hour * 60 + wake_dt.minute
|
minutes = wake_dt.hour * 60 + wake_dt.minute
|
||||||
floored = minutes - (minutes % 5)
|
floored = minutes - (minutes % 5)
|
||||||
return [tf for tf in DASHBOARD_TFS if floored % TF_INTERVAL_MINUTES[tf] == 0]
|
boundary = wake_dt.replace(
|
||||||
|
hour=floored // 60, minute=floored % 60, second=0, microsecond=0)
|
||||||
|
return [
|
||||||
|
tf for tf in DASHBOARD_TFS
|
||||||
|
if floored % TF_INTERVAL_MINUTES[tf] == 0
|
||||||
|
and not _providers_base._is_daily_maintenance(
|
||||||
|
boundary - timedelta(minutes=TF_INTERVAL_MINUTES[tf]))
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def catch_up_timeframes(wake_dt: datetime,
|
||||||
|
last_boundary: Dict[str, float]) -> List[Timeframe]:
|
||||||
|
"""Timeframes whose most recent closed boundary (<= wake) is newer than
|
||||||
|
the boundary of their last successful scan — i.e. boundaries that were
|
||||||
|
missed while a cycle overran or the process was down. A TF never scanned
|
||||||
|
is due. Excludes bars whose label sits in the daily settlement pause
|
||||||
|
(same rule as due_timeframes).
|
||||||
|
|
||||||
|
`due_timeframes` only ever names the timeframes whose boundary coincides
|
||||||
|
with THIS wake, so an overrunning cycle (scan + advisory panel + news can
|
||||||
|
take minutes) or a restart left the page serving the previous H1/H4
|
||||||
|
report for up to 1 h / 4 h while advertising a fresh next_close_epoch
|
||||||
|
(2026-09-13 audit)."""
|
||||||
|
minutes = wake_dt.hour * 60 + wake_dt.minute
|
||||||
|
out = []
|
||||||
|
for tf in DASHBOARD_TFS:
|
||||||
|
interval = TF_INTERVAL_MINUTES[tf]
|
||||||
|
floored = minutes - (minutes % interval)
|
||||||
|
last_closed = wake_dt.replace(
|
||||||
|
hour=floored // 60, minute=floored % 60, second=0, microsecond=0)
|
||||||
|
# No bar carries a label inside a scheduled closure (settlement
|
||||||
|
# pause or weekend), so nothing can have been "missed" there — and a
|
||||||
|
# weekend-blocked scan never advances the ledger, so without the
|
||||||
|
# weekend half of this test every weekend wake would re-scan
|
||||||
|
# M15/H1/H4 for nothing (~1,800 blocked rows per weekend).
|
||||||
|
# (Whole-span rule, as in providers.base._has_latest_closed_bar: an
|
||||||
|
# H4 bar labelled Sunday 20:00 UTC starts in the weekend but runs
|
||||||
|
# through the reopen, so it exists.)
|
||||||
|
label = last_closed - timedelta(minutes=interval)
|
||||||
|
if (_providers_base._is_scheduled_closure(label)
|
||||||
|
and _providers_base._is_scheduled_closure(
|
||||||
|
last_closed - timedelta(seconds=1))):
|
||||||
|
continue
|
||||||
|
scanned = last_boundary.get(tf.value)
|
||||||
|
if scanned is None or scanned < last_closed.timestamp():
|
||||||
|
out.append(tf)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
||||||
@@ -324,6 +381,9 @@ class DashboardPublisher:
|
|||||||
self.out_dir = out_dir
|
self.out_dir = out_dir
|
||||||
self.snapshots: Dict[str, dict] = {}
|
self.snapshots: Dict[str, dict] = {}
|
||||||
self._reports = {}
|
self._reports = {}
|
||||||
|
# tf.value → epoch seconds of the boundary the last successful scan
|
||||||
|
# belonged to; feeds catch_up_timeframes (2026-09-13 audit).
|
||||||
|
self._last_boundary: Dict[str, float] = {}
|
||||||
self.news_payload: Optional[dict] = None
|
self.news_payload: Optional[dict] = None
|
||||||
self._dp = None
|
self._dp = None
|
||||||
self._news_engine: Optional[NewsEngine] = None
|
self._news_engine: Optional[NewsEngine] = None
|
||||||
@@ -355,7 +415,14 @@ class DashboardPublisher:
|
|||||||
shutil.copyfile(src, dst + ".tmp")
|
shutil.copyfile(src, dst + ".tmp")
|
||||||
os.replace(dst + ".tmp", dst)
|
os.replace(dst + ".tmp", dst)
|
||||||
|
|
||||||
def scan_tf(self, tf: Timeframe) -> None:
|
def scan_tf(self, tf: Timeframe,
|
||||||
|
boundary_epoch: Optional[float] = None) -> None:
|
||||||
|
"""Scan one timeframe and store its snapshot entry.
|
||||||
|
|
||||||
|
`boundary_epoch` is the fallback boundary recorded when the report
|
||||||
|
carries no analysis (weekend/blocked or failed scan) or its bar
|
||||||
|
label will not parse; normally the boundary comes from the report
|
||||||
|
itself (bar label + interval)."""
|
||||||
started = time.time()
|
started = time.time()
|
||||||
try:
|
try:
|
||||||
report = produce_signal(XAUUSD, tf, self._provider(),
|
report = produce_signal(XAUUSD, tf, self._provider(),
|
||||||
@@ -365,6 +432,19 @@ class DashboardPublisher:
|
|||||||
if violent_watch_enabled():
|
if violent_watch_enabled():
|
||||||
attach_watch(report, tf,
|
attach_watch(report, tf,
|
||||||
get_app_config().violent_watch_settings)
|
get_app_config().violent_watch_settings)
|
||||||
|
# Which boundary this report belongs to: the bar LABEL it closed
|
||||||
|
# on plus one interval. Recorded so a boundary missed by an
|
||||||
|
# overrunning cycle or a restart is caught up on the next wake
|
||||||
|
# instead of sitting stale (2026-09-13 audit).
|
||||||
|
boundary = boundary_epoch
|
||||||
|
if report.primary_analysis is not None:
|
||||||
|
try:
|
||||||
|
boundary = (pd.Timestamp(report.timestamp).timestamp()
|
||||||
|
+ TF_INTERVAL_MINUTES[tf] * 60)
|
||||||
|
except Exception:
|
||||||
|
boundary = boundary_epoch
|
||||||
|
if boundary is not None:
|
||||||
|
self._last_boundary[tf.value] = boundary
|
||||||
self._reports[tf.value] = report
|
self._reports[tf.value] = report
|
||||||
self.snapshots[tf.value] = {
|
self.snapshots[tf.value] = {
|
||||||
"status": "ok",
|
"status": "ok",
|
||||||
@@ -449,9 +529,15 @@ class DashboardPublisher:
|
|||||||
logger.info(f"next M5 boundary in {delay:.0f}s")
|
logger.info(f"next M5 boundary in {delay:.0f}s")
|
||||||
time.sleep(delay)
|
time.sleep(delay)
|
||||||
try:
|
try:
|
||||||
due = due_timeframes(datetime.now(timezone.utc))
|
now = datetime.now(timezone.utc)
|
||||||
|
due = due_timeframes(now)
|
||||||
logger.info(f"boundary wake — scanning {[t.value for t in due]}")
|
logger.info(f"boundary wake — scanning {[t.value for t in due]}")
|
||||||
self.run_cycle(due)
|
missed = catch_up_timeframes(now, self._last_boundary)
|
||||||
|
extra = [tf for tf in missed if tf not in due]
|
||||||
|
if extra:
|
||||||
|
logger.info(f"catch-up — scanning "
|
||||||
|
f"{[t.value for t in extra]} (missed boundary)")
|
||||||
|
self.run_cycle(due + extra)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("cycle failed")
|
logger.exception("cycle failed")
|
||||||
|
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ Design philosophy:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
import math
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from typing import Dict, List, Optional, Tuple
|
from typing import Dict, List, Optional, Tuple
|
||||||
|
|
||||||
@@ -84,15 +85,65 @@ _DEFAULT_REGIME_WEIGHTS = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_regime_override(override: Dict[str, Dict[str, float]]) -> None:
|
||||||
|
"""Reject a PARTIAL ``regime_weights_override`` (raises ValueError).
|
||||||
|
|
||||||
|
An override REPLACES the whole table, it does not patch it: whatever
|
||||||
|
it omits is multiplied by 1.0 in _compute_composite, not by the
|
||||||
|
shipped default. A half-filled research table (``--set
|
||||||
|
technical.regime_weights_override=<JSON>``) would therefore silently
|
||||||
|
run 1.0 where the default said 0.3 or 2.5, with no warning — so every
|
||||||
|
regime and all nine scored indicators must be present, with a finite
|
||||||
|
multiplier >= 0. The shipped M15 override is derived from the default
|
||||||
|
table (pair_signals/xauusd._m15_regime_weights) and passes."""
|
||||||
|
problems: List[str] = []
|
||||||
|
want_regimes = set(_DEFAULT_REGIME_WEIGHTS)
|
||||||
|
want_indicators = set(_TRENDING_WEIGHTS)
|
||||||
|
got_regimes = set(override)
|
||||||
|
if got_regimes != want_regimes:
|
||||||
|
problems.append(
|
||||||
|
f"missing regimes {sorted(want_regimes - got_regimes)}, "
|
||||||
|
f"unexpected {sorted(got_regimes - want_regimes)}")
|
||||||
|
for regime in sorted(got_regimes & want_regimes):
|
||||||
|
weights = override[regime]
|
||||||
|
if not isinstance(weights, dict):
|
||||||
|
problems.append(f"regime '{regime}' is not an indicator -> "
|
||||||
|
f"multiplier mapping: {weights!r}")
|
||||||
|
continue
|
||||||
|
got = set(weights)
|
||||||
|
if got != want_indicators:
|
||||||
|
problems.append(
|
||||||
|
f"regime '{regime}' missing {sorted(want_indicators - got)}, "
|
||||||
|
f"unexpected {sorted(got - want_indicators)}")
|
||||||
|
for name in sorted(got & want_indicators):
|
||||||
|
mult = weights[name]
|
||||||
|
if (isinstance(mult, bool) or not isinstance(mult, (int, float))
|
||||||
|
or not math.isfinite(mult) or mult < 0):
|
||||||
|
problems.append(
|
||||||
|
f"regime '{regime}' indicator '{name}' multiplier "
|
||||||
|
f"{mult!r} is not a finite number >= 0")
|
||||||
|
if problems:
|
||||||
|
raise ValueError(
|
||||||
|
"regime_weights_override is partial: " + "; ".join(problems)
|
||||||
|
+ "; an override replaces the WHOLE table "
|
||||||
|
"(see engine._DEFAULT_REGIME_WEIGHTS)")
|
||||||
|
|
||||||
|
|
||||||
def regime_weight_table(config: TechnicalConfig) -> Dict[str, Dict[str, float]]:
|
def regime_weight_table(config: TechnicalConfig) -> Dict[str, Dict[str, float]]:
|
||||||
"""The regime -> indicator weight multipliers _compute_composite runs
|
"""The regime -> indicator weight multipliers _compute_composite runs
|
||||||
with (``regime_weights_override`` wins, else the gold-optimized
|
with (``regime_weights_override`` wins, else the gold-optimized
|
||||||
defaults above).
|
defaults above).
|
||||||
|
|
||||||
|
A non-empty override must cover the WHOLE table (every regime, all
|
||||||
|
nine scored indicators) — a partial one raises ValueError instead of
|
||||||
|
silently multiplying the cells it omits by 1.0.
|
||||||
|
|
||||||
Pure and side-effect free: ONE source of truth shared by the composite
|
Pure and side-effect free: ONE source of truth shared by the composite
|
||||||
and by signal_aggregator.engine_config_fingerprint. Returns a copy, so
|
and by signal_aggregator.engine_config_fingerprint. Returns a copy, so
|
||||||
a caller can never mutate the shipped table."""
|
a caller can never mutate the shipped table."""
|
||||||
override = getattr(config, "regime_weights_override", None)
|
override = getattr(config, "regime_weights_override", None)
|
||||||
|
if override:
|
||||||
|
_validate_regime_override(override)
|
||||||
table = override if override else _DEFAULT_REGIME_WEIGHTS
|
table = override if override else _DEFAULT_REGIME_WEIGHTS
|
||||||
return {regime: dict(weights) for regime, weights in table.items()}
|
return {regime: dict(weights) for regime, weights in table.items()}
|
||||||
|
|
||||||
@@ -119,7 +170,7 @@ class TechnicalEngine:
|
|||||||
return TechnicalAnalysis(
|
return TechnicalAnalysis(
|
||||||
symbol=symbol,
|
symbol=symbol,
|
||||||
timeframe=timeframe,
|
timeframe=timeframe,
|
||||||
timestamp=str(pd.Timestamp.utcnow()),
|
timestamp=str(pd.Timestamp.now(tz="UTC")),
|
||||||
confidence=0.0,
|
confidence=0.0,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
+122
-20
@@ -65,6 +65,46 @@ def _is_forex_weekend(now=None) -> bool:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _is_daily_maintenance(now=None) -> bool:
|
||||||
|
"""Gold's daily settlement pause: Monday-Thursday 17:00-17:59 New York
|
||||||
|
(21:00 UTC in US summer, 22:00 in winter). No bar carries a label
|
||||||
|
inside this hour. Friday 17:00 is the weekend (see _is_forex_weekend).
|
||||||
|
Same rule as advisory_outcome._has_continuity's daily_maintenance."""
|
||||||
|
if now is None:
|
||||||
|
now = pd.Timestamp.now(tz="UTC")
|
||||||
|
ny = now.astimezone(_NY_TZ)
|
||||||
|
return ny.weekday() < 4 and ny.hour == 17
|
||||||
|
|
||||||
|
|
||||||
|
def _is_scheduled_closure(now=None) -> bool:
|
||||||
|
"""The feed is legitimately silent: weekend or daily settlement pause."""
|
||||||
|
return _is_forex_weekend(now) or _is_daily_maintenance(now)
|
||||||
|
|
||||||
|
|
||||||
|
def _scheduled_closed_seconds(start, end) -> float:
|
||||||
|
"""Seconds of the [start, end) span that fall inside a scheduled
|
||||||
|
closure (weekend or the daily settlement pause).
|
||||||
|
|
||||||
|
Bar age is measured from a bar LABEL, so at 21:30 UTC in summer the
|
||||||
|
20:55 M5 bar looks 35 minutes old even though the feed could not have
|
||||||
|
printed anything since 21:00. Subtracting the closed span keeps the
|
||||||
|
3-candle freshness limit meaningful instead of failing it — and
|
||||||
|
logging a false "feed lagging"/"STALE DATA" — every settlement hour
|
||||||
|
(2026-09-13 audit: 61 suppressed scans on three nights).
|
||||||
|
"""
|
||||||
|
hour = pd.Timedelta(hours=1)
|
||||||
|
closed = 0.0
|
||||||
|
t = start.floor("h")
|
||||||
|
steps = 0
|
||||||
|
while t < end and steps < 24 * 10:
|
||||||
|
if _is_scheduled_closure(t):
|
||||||
|
closed += max(
|
||||||
|
0.0, (min(t + hour, end) - max(t, start)).total_seconds())
|
||||||
|
t += hour
|
||||||
|
steps += 1
|
||||||
|
return closed
|
||||||
|
|
||||||
|
|
||||||
class DataIntegrityError(Exception):
|
class DataIntegrityError(Exception):
|
||||||
"""Raised when data fails validation checks."""
|
"""Raised when data fails validation checks."""
|
||||||
pass
|
pass
|
||||||
@@ -276,7 +316,7 @@ class BaseDataProvider(ABC):
|
|||||||
chunks = []
|
chunks = []
|
||||||
cursor_end = end
|
cursor_end = end
|
||||||
empty_streak = 0
|
empty_streak = 0
|
||||||
error_streak = 0
|
window_errors = 0
|
||||||
while cursor_end > start:
|
while cursor_end > start:
|
||||||
cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
|
cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
|
||||||
try:
|
try:
|
||||||
@@ -286,23 +326,31 @@ class BaseDataProvider(ABC):
|
|||||||
)
|
)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
# A raising chunk is a PROVIDER ERROR (auth, 429, 5xx,
|
# A raising chunk is a PROVIDER ERROR (auth, 429, 5xx,
|
||||||
# malformed payload) — not "no data". Three in a row means
|
# malformed payload) — not "no data". RETRY THE SAME
|
||||||
# the rest of the history is unreachable: raise instead of
|
# WINDOW: until 2026-09-13 a failure advanced the cursor
|
||||||
# returning a silently truncated frame (the backtester
|
# past the window, so one or two NON-consecutive page
|
||||||
# would treat it as complete).
|
# failures (Oanda's sporadic 401/5xx, even after the
|
||||||
|
# provider's own retry ladder) silently dropped up to
|
||||||
|
# days_per_chunk of bars — 15 days of M5 — out of the
|
||||||
|
# middle of the frame, and only the HEAD was
|
||||||
|
# coverage-checked below, so the backtester cached the hole
|
||||||
|
# as complete history forever. Three failures on the SAME
|
||||||
|
# window means it is unreachable: raise instead of
|
||||||
|
# returning a frame with a hole in it.
|
||||||
logger.warning(
|
logger.warning(
|
||||||
f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
|
f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
|
||||||
)
|
)
|
||||||
error_streak += 1
|
window_errors += 1
|
||||||
if error_streak >= 3:
|
if window_errors >= 3:
|
||||||
raise self.DATA_ERROR(
|
raise self.DATA_ERROR(
|
||||||
f"{symbol} {timeframe.value}: {error_streak} "
|
f"{symbol} {timeframe.value}: {window_errors} "
|
||||||
f"consecutive chunk failures walking back from "
|
f"consecutive failures fetching window "
|
||||||
f"{cursor_end.date()} — refusing to return "
|
f"{cursor_start.date()}–{cursor_end.date()} — "
|
||||||
f"truncated history (last error: {e})") from e
|
f"refusing to return truncated history "
|
||||||
cursor_end = cursor_start - timedelta(seconds=1)
|
f"(last error: {e})") from e
|
||||||
continue
|
continue # re-request the SAME window
|
||||||
|
|
||||||
|
window_errors = 0
|
||||||
if chunk is None or chunk.empty:
|
if chunk is None or chunk.empty:
|
||||||
# Genuinely empty windows (weekends/holidays/asset gaps)
|
# Genuinely empty windows (weekends/holidays/asset gaps)
|
||||||
# are normal; several in a row = start of history.
|
# are normal; several in a row = start of history.
|
||||||
@@ -313,7 +361,6 @@ class BaseDataProvider(ABC):
|
|||||||
continue
|
continue
|
||||||
|
|
||||||
empty_streak = 0
|
empty_streak = 0
|
||||||
error_streak = 0
|
|
||||||
chunks.append(chunk)
|
chunks.append(chunk)
|
||||||
# Step back; dedup on concat.
|
# Step back; dedup on concat.
|
||||||
new_cursor = chunk.index[0] - timedelta(seconds=1)
|
new_cursor = chunk.index[0] - timedelta(seconds=1)
|
||||||
@@ -348,6 +395,38 @@ class BaseDataProvider(ABC):
|
|||||||
f"{df.index[0]} but {start_ts} was requested — "
|
f"{df.index[0]} but {start_ts} was requested — "
|
||||||
f"~{(df.index[0] - start_ts).days} days missing from the "
|
f"~{(df.index[0] - start_ts).days} days missing from the "
|
||||||
f"head; treat downstream results accordingly")
|
f"head; treat downstream results accordingly")
|
||||||
|
# INTERIOR-gap audit (2026-09-13): the head check above is blind to
|
||||||
|
# a hole in the MIDDLE of the frame, which is exactly what a
|
||||||
|
# dropped page used to leave behind. Missing labels inside a
|
||||||
|
# scheduled closure (weekend, daily settlement pause) are normal,
|
||||||
|
# so only unscheduled runs are reported — and a real market holiday
|
||||||
|
# (Christmas, Good Friday, July 4) looks identical to a lost page,
|
||||||
|
# so this is LOUD BUT NOT FATAL and the frame is returned unchanged.
|
||||||
|
if len(df) > 400_000:
|
||||||
|
logger.info(
|
||||||
|
f"{symbol} {timeframe.value}: interior-gap audit skipped "
|
||||||
|
f"({len(df)} rows)")
|
||||||
|
elif len(df) > 1:
|
||||||
|
step = pd.Timedelta(seconds=self.CANDLE_SECONDS[timeframe])
|
||||||
|
labels = df.index
|
||||||
|
# Missing labels between each consecutive pair; >3 is the floor
|
||||||
|
# worth a look (weekends/settlement hours leave 1-2 labels that
|
||||||
|
# the NY-anchored closure rule does not cover).
|
||||||
|
spans = ((labels[1:] - labels[:-1]) / step).astype("int64") - 1
|
||||||
|
gaps = []
|
||||||
|
for i in (spans > 3).nonzero()[0]:
|
||||||
|
missing = sum(
|
||||||
|
1 for k in range(1, int(spans[i]) + 1)
|
||||||
|
if not _is_scheduled_closure(labels[i] + k * step))
|
||||||
|
if missing > 3:
|
||||||
|
gaps.append((labels[i], labels[i + 1], missing))
|
||||||
|
if gaps:
|
||||||
|
logger.error(
|
||||||
|
f"{symbol} {timeframe.value}: {len(gaps)} interior "
|
||||||
|
f"gap(s) of >3 unscheduled bars in fetched history — "
|
||||||
|
f"first: {gaps[0]}; a page fetch may have been dropped "
|
||||||
|
f"or the market was closed (holiday); verify before "
|
||||||
|
f"trusting this window")
|
||||||
return df
|
return df
|
||||||
|
|
||||||
def is_data_fresh(
|
def is_data_fresh(
|
||||||
@@ -398,7 +477,8 @@ class BaseDataProvider(ABC):
|
|||||||
|
|
||||||
# ── Shared internals ───────────────────────────────────────────
|
# ── Shared internals ───────────────────────────────────────────
|
||||||
|
|
||||||
def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool:
|
def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe,
|
||||||
|
now=None) -> bool:
|
||||||
"""Cached data must include the most recent fully closed bar.
|
"""Cached data must include the most recent fully closed bar.
|
||||||
|
|
||||||
The TTL alone is wall-clock based: a snapshot fetched mid-candle is
|
The TTL alone is wall-clock based: a snapshot fetched mid-candle is
|
||||||
@@ -408,12 +488,29 @@ class BaseDataProvider(ABC):
|
|||||||
"""
|
"""
|
||||||
if df is None or df.empty:
|
if df is None or df.empty:
|
||||||
return False
|
return False
|
||||||
now = pd.Timestamp.now(tz="UTC")
|
if now is None:
|
||||||
|
now = pd.Timestamp.now(tz="UTC")
|
||||||
if _is_forex_weekend(now):
|
if _is_forex_weekend(now):
|
||||||
return True
|
return True
|
||||||
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
|
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
|
||||||
dur = pd.Timedelta(seconds=candle_seconds)
|
dur = pd.Timedelta(seconds=candle_seconds)
|
||||||
expected_label = now.floor(dur) - dur
|
expected_label = now.floor(dur) - dur
|
||||||
|
# No bar ever carries a label inside a scheduled closure, so the
|
||||||
|
# newest bar that MUST be present is the last label before it.
|
||||||
|
# Without this walk-back every scan from 21:05 to 22:00 UTC in
|
||||||
|
# summer evicted the cache, refetched twice and logged a false
|
||||||
|
# "feed lagging" (2026-09-13 audit).
|
||||||
|
# A label counts as "inside" only when the closure covers the bar's
|
||||||
|
# whole span: a D1 bar labelled Sunday 00:00 UTC starts in the
|
||||||
|
# weekend but runs through the Sunday reopen, so it exists and
|
||||||
|
# must be present (68 such bars in the 3-year Oanda D1 cache).
|
||||||
|
steps = 0
|
||||||
|
last_second = dur - pd.Timedelta(seconds=1)
|
||||||
|
while (_is_scheduled_closure(expected_label)
|
||||||
|
and _is_scheduled_closure(expected_label + last_second)
|
||||||
|
and steps < 2000):
|
||||||
|
expected_label -= dur
|
||||||
|
steps += 1
|
||||||
return df.index[-1] >= expected_label
|
return df.index[-1] >= expected_label
|
||||||
|
|
||||||
def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame:
|
def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame:
|
||||||
@@ -443,12 +540,14 @@ class BaseDataProvider(ABC):
|
|||||||
timeframe.value, dropped)
|
timeframe.value, dropped)
|
||||||
return df
|
return df
|
||||||
|
|
||||||
def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool:
|
def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe,
|
||||||
|
now=None) -> bool:
|
||||||
"""Check if the last candle is within acceptable age for this timeframe."""
|
"""Check if the last candle is within acceptable age for this timeframe."""
|
||||||
if df is None or df.empty:
|
if df is None or df.empty:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
now = pd.Timestamp.now(tz="UTC")
|
if now is None:
|
||||||
|
now = pd.Timestamp.now(tz="UTC")
|
||||||
|
|
||||||
if _is_forex_weekend(now):
|
if _is_forex_weekend(now):
|
||||||
return True
|
return True
|
||||||
@@ -456,7 +555,10 @@ class BaseDataProvider(ABC):
|
|||||||
last_bar = df.index[-1]
|
last_bar = df.index[-1]
|
||||||
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
|
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
|
||||||
max_age_seconds = candle_seconds * 3
|
max_age_seconds = candle_seconds * 3
|
||||||
age_seconds = (now - last_bar).total_seconds()
|
# Scheduled closures (weekend, daily settlement pause) are dead
|
||||||
|
# time in which no bar can print — they must not count as age.
|
||||||
|
age_seconds = ((now - last_bar).total_seconds()
|
||||||
|
- _scheduled_closed_seconds(last_bar, now))
|
||||||
|
|
||||||
if age_seconds > max_age_seconds:
|
if age_seconds > max_age_seconds:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
@@ -482,7 +584,7 @@ class BaseDataProvider(ABC):
|
|||||||
df = df.sort_index()
|
df = df.sort_index()
|
||||||
df = df[~df.index.duplicated(keep="last")]
|
df = df[~df.index.duplicated(keep="last")]
|
||||||
|
|
||||||
now = pd.Timestamp.utcnow()
|
now = pd.Timestamp.now(tz="UTC")
|
||||||
future_mask = df.index > now + pd.Timedelta(minutes=5)
|
future_mask = df.index > now + pd.Timedelta(minutes=5)
|
||||||
if future_mask.any():
|
if future_mask.any():
|
||||||
logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps")
|
logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps")
|
||||||
|
|||||||
+2
-2
@@ -1,5 +1,5 @@
|
|||||||
numpy>=1.24
|
numpy>=1.24,<3
|
||||||
pandas>=2.0
|
pandas>=2.0,<4
|
||||||
requests>=2.28
|
requests>=2.28
|
||||||
twelvedata>=1.2.0
|
twelvedata>=1.2.0
|
||||||
python-telegram-bot>=20.0
|
python-telegram-bot>=20.0
|
||||||
|
|||||||
+62
-8
@@ -242,6 +242,36 @@ class SignalAggregator:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}")
|
logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}")
|
||||||
|
|
||||||
|
# A missing frame is a FAILED scan, never a degraded one. Losing a
|
||||||
|
# CONFIRMATION frame used to let the scan continue with fewer
|
||||||
|
# frames: _check_mtf_agreement needs >= 2 signals, so it returned
|
||||||
|
# False and the composite took a 15-point mtf_disagreement penalty
|
||||||
|
# the market never earned (A/A+ became impossible), while on
|
||||||
|
# M15/H1/H4 a B/C setup could still ship on one confirmation with
|
||||||
|
# no warning at all. The backtester never sees that state
|
||||||
|
# (run_backtest raises on an empty frame), so it was a live-only
|
||||||
|
# divergence. 2026-09-13 audit.
|
||||||
|
missing = [tf for tf in timeframes if tf not in dfs]
|
||||||
|
if missing:
|
||||||
|
report = SignalReport(
|
||||||
|
symbol=symbol,
|
||||||
|
display_name=pair_info["display"],
|
||||||
|
timestamp=str(now_utc),
|
||||||
|
)
|
||||||
|
report.final_signal = SignalStrength.NEUTRAL
|
||||||
|
report.signal_quality = "F"
|
||||||
|
report.trade_setup = None
|
||||||
|
if self.config.primary_timeframe in missing:
|
||||||
|
# Legacy text signal_log.gate_reason already recognises.
|
||||||
|
report.warnings.append("Primary timeframe data unavailable")
|
||||||
|
report.warnings.append(
|
||||||
|
f"DATA UNAVAILABLE: {', '.join(tf.value for tf in missing)}"
|
||||||
|
" — scan failed"
|
||||||
|
)
|
||||||
|
report.action_recommendation = "NO TRADE — market data unavailable"
|
||||||
|
self._log_signal(report, log_source, dfs, failed=True)
|
||||||
|
return report
|
||||||
|
|
||||||
report = self.analyze_from_dataframes(
|
report = self.analyze_from_dataframes(
|
||||||
symbol=symbol,
|
symbol=symbol,
|
||||||
dfs=dfs,
|
dfs=dfs,
|
||||||
@@ -303,7 +333,7 @@ class SignalAggregator:
|
|||||||
def _log_signal(
|
def _log_signal(
|
||||||
self, report: SignalReport, source: Optional[str],
|
self, report: SignalReport, source: Optional[str],
|
||||||
dataframes: Optional[Dict[Timeframe, pd.DataFrame]] = None,
|
dataframes: Optional[Dict[Timeframe, pd.DataFrame]] = None,
|
||||||
blocked: bool = False,
|
blocked: bool = False, failed: bool = False,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Persist the scan to the signal-history DB (signal_log.py).
|
"""Persist the scan to the signal-history DB (signal_log.py).
|
||||||
|
|
||||||
@@ -314,9 +344,11 @@ class SignalAggregator:
|
|||||||
is still written, just without the news JSON blobs. ``blocked``
|
is still written, just without the news JSON blobs. ``blocked``
|
||||||
marks weekend/day early returns: no news fetch, no snapshots —
|
marks weekend/day early returns: no news fetch, no snapshots —
|
||||||
a ~1 KB liveness row (or nothing, per signal_log.log_blocked_scans).
|
a ~1 KB liveness row (or nothing, per signal_log.log_blocked_scans).
|
||||||
A scan whose PRIMARY fetch failed is logged as ``scan_kind``
|
A scan missing ANY required frame (primary or confirmation, or
|
||||||
``failed`` (price 0, no signal): liveness evidence that readers
|
the regime-filter frame when that filter is on) is logged as
|
||||||
skip like ``blocked``, so it never pollutes NEUTRAL statistics."""
|
``scan_kind`` ``failed`` (price 0, no signal): liveness evidence
|
||||||
|
that readers skip like ``blocked``, so it never pollutes NEUTRAL
|
||||||
|
statistics."""
|
||||||
context = None
|
context = None
|
||||||
if not blocked:
|
if not blocked:
|
||||||
try:
|
try:
|
||||||
@@ -325,10 +357,10 @@ class SignalAggregator:
|
|||||||
logger.debug("signal log: news context unavailable",
|
logger.debug("signal log: news context unavailable",
|
||||||
exc_info=True)
|
exc_info=True)
|
||||||
kind = "blocked" if blocked else "scan"
|
kind = "blocked" if blocked else "scan"
|
||||||
if not blocked and dataframes is not None:
|
primary = (dataframes or {}).get(self.config.primary_timeframe)
|
||||||
primary = dataframes.get(self.config.primary_timeframe)
|
if failed or (not blocked and dataframes is not None
|
||||||
if primary is None or primary.empty:
|
and (primary is None or primary.empty)):
|
||||||
kind = "failed"
|
kind = "failed"
|
||||||
extra = {"scan_kind": kind}
|
extra = {"scan_kind": kind}
|
||||||
for key, fingerprint in (
|
for key, fingerprint in (
|
||||||
("config_fingerprint", self.gate_config_fingerprint),
|
("config_fingerprint", self.gate_config_fingerprint),
|
||||||
@@ -1297,6 +1329,28 @@ class SignalAggregator:
|
|||||||
report.trade_setup = self._generate_xau_atr_trade_setup(report, atr)
|
report.trade_setup = self._generate_xau_atr_trade_setup(report, atr)
|
||||||
if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"):
|
if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"):
|
||||||
report.trade_setup = self._generate_trade_setup(report, pip_value)
|
report.trade_setup = self._generate_trade_setup(report, pip_value)
|
||||||
|
# A shipped setup always names a direction. The signal thresholds
|
||||||
|
# (BUY/SELL need confidence >= 50) sit above the C grade floor
|
||||||
|
# (>= 45), and gate 6c grants B at >= 45 in "transitioning", so a
|
||||||
|
# C/B report could leave here as NEUTRAL with a full setup: the
|
||||||
|
# backtester traded it (the replay never reads final_signal) while
|
||||||
|
# the Telegram alert gate rejects NEUTRAL first (2026-09-13 audit:
|
||||||
|
# 5/91 H1 12-mo book trades, +3.7R of +22.2R, were never alerted).
|
||||||
|
# The generators derived the direction from the score sign — the
|
||||||
|
# same sign every direction gate above used — so this changes
|
||||||
|
# display and alert eligibility only; the trade books are unchanged.
|
||||||
|
# A report no generator served stays NEUTRAL, as before.
|
||||||
|
if (report.trade_setup is not None
|
||||||
|
and report.final_signal == SignalStrength.NEUTRAL):
|
||||||
|
report.final_signal = (SignalStrength.BUY
|
||||||
|
if report.trade_setup.direction == "BUY"
|
||||||
|
else SignalStrength.SELL)
|
||||||
|
# The action text was written before the gate chain with the
|
||||||
|
# old signal name; keep it consistent with the header.
|
||||||
|
if report.action_recommendation.startswith("NEUTRAL"):
|
||||||
|
report.action_recommendation = (
|
||||||
|
report.final_signal.name.replace("_", " ")
|
||||||
|
+ report.action_recommendation[len("NEUTRAL"):])
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _drop_stale_spread_gate_vetoes(report: SignalReport) -> None:
|
def _drop_stale_spread_gate_vetoes(report: SignalReport) -> None:
|
||||||
|
|||||||
+127
-16
@@ -104,6 +104,7 @@ from advisory_outcome import (
|
|||||||
canonical_r,
|
canonical_r,
|
||||||
initial_outcome,
|
initial_outcome,
|
||||||
normalize_timeframe,
|
normalize_timeframe,
|
||||||
|
resolve_locked_floor,
|
||||||
timeframe_value,
|
timeframe_value,
|
||||||
)
|
)
|
||||||
# Private on purpose — the scan-outcome labeller must apply EXACTLY the
|
# Private on purpose — the scan-outcome labeller must apply EXACTLY the
|
||||||
@@ -258,6 +259,7 @@ _GATE_PREFIXES = (
|
|||||||
"SESSION FILTER",
|
"SESSION FILTER",
|
||||||
"Primary timeframe stale",
|
"Primary timeframe stale",
|
||||||
"Primary timeframe data unavailable",
|
"Primary timeframe data unavailable",
|
||||||
|
"DATA UNAVAILABLE:",
|
||||||
"REGIME FILTER",
|
"REGIME FILTER",
|
||||||
"EXHAUSTION BLOCK",
|
"EXHAUSTION BLOCK",
|
||||||
"EXHAUSTED TREND FILTER",
|
"EXHAUSTED TREND FILTER",
|
||||||
@@ -278,6 +280,10 @@ SCAN_OUTCOME_WINDOW_H = 24
|
|||||||
# A 24 h window can only be judged once it has elapsed; the extra hour
|
# A 24 h window can only be judged once it has elapsed; the extra hour
|
||||||
# leaves room for the last bar of the window to be stored.
|
# leaves room for the last bar of the window to be stored.
|
||||||
SCAN_LABEL_MIN_AGE_H = 25
|
SCAN_LABEL_MIN_AGE_H = 25
|
||||||
|
# An incomplete row older than this is FINAL: its window had a real hole
|
||||||
|
# (a feed outage), which no later bar can fill. It is neither re-examined
|
||||||
|
# nor allowed to anchor the shared bar stream for the rest of the batch.
|
||||||
|
SCAN_LABEL_MAX_AGE_H = 72
|
||||||
# Missing bars tolerated inside a window ON TOP of the scheduled closures
|
# Missing bars tolerated inside a window ON TOP of the scheduled closures
|
||||||
# advisory_outcome._has_continuity already allows.
|
# advisory_outcome._has_continuity already allows.
|
||||||
SCAN_LABEL_MAX_GAP_BARS = 3
|
SCAN_LABEL_MAX_GAP_BARS = 3
|
||||||
@@ -399,6 +405,20 @@ def setup_key(setup: Optional[Dict[str, Any]], timeframe: Any) -> Optional[str]:
|
|||||||
))
|
))
|
||||||
|
|
||||||
|
|
||||||
|
def _is_gap_reason(reason: Any) -> bool:
|
||||||
|
"""True when an outcome was left UNRESOLVABLE by a candle gap.
|
||||||
|
|
||||||
|
advance_outcome writes TWO such reasons — "stored candle snapshots
|
||||||
|
lost continuity" (no overlap with the last processed bar) and "stored
|
||||||
|
candle snapshots contain an internal gap" — and each call site used to
|
||||||
|
match only one of them, so the stream rebuild and the locked-floor
|
||||||
|
rescue skipped half the gap rows. The substring clauses keep states
|
||||||
|
written by older code inside the same net."""
|
||||||
|
text = str(reason or "")
|
||||||
|
return (text.startswith("stored candle snapshots")
|
||||||
|
or "continuity" in text or "gap" in text)
|
||||||
|
|
||||||
|
|
||||||
def gate_reason(
|
def gate_reason(
|
||||||
warnings: Iterable[str],
|
warnings: Iterable[str],
|
||||||
cleaned_warnings: Iterable[str],
|
cleaned_warnings: Iterable[str],
|
||||||
@@ -1150,12 +1170,19 @@ class SignalLogger:
|
|||||||
before = _dumps(json.loads(hit[0]))
|
before = _dumps(json.loads(hit[0]))
|
||||||
advanced = advance_outcome(json.loads(before), bars)
|
advanced = advance_outcome(json.loads(before), bars)
|
||||||
if (advanced.get("status") == "UNRESOLVABLE"
|
if (advanced.get("status") == "UNRESOLVABLE"
|
||||||
and "continuity" in str(advanced.get("reason", ""))):
|
and _is_gap_reason(advanced.get("reason"))):
|
||||||
union = self._stream_since(
|
union = self._stream_since(
|
||||||
symbol, timeframe,
|
symbol, timeframe,
|
||||||
json.loads(before)["progress"]["last_processed_bar"])
|
json.loads(before)["progress"]["last_processed_bar"])
|
||||||
if union:
|
if union:
|
||||||
advanced = advance_outcome(json.loads(before), union)
|
advanced = advance_outcome(json.loads(before), union)
|
||||||
|
# The database could not fill the hole. A trade
|
||||||
|
# whose trail had already locked profit still has a
|
||||||
|
# real answer — its floor — so record that instead
|
||||||
|
# of dropping the row out of the scorecard.
|
||||||
|
if (advanced.get("status") == "UNRESOLVABLE"
|
||||||
|
and _is_gap_reason(advanced.get("reason"))):
|
||||||
|
advanced = resolve_locked_floor(advanced)
|
||||||
after = _dumps(advanced)
|
after = _dumps(advanced)
|
||||||
if after == before:
|
if after == before:
|
||||||
self._conn.rollback()
|
self._conn.rollback()
|
||||||
@@ -1194,6 +1221,11 @@ class SignalLogger:
|
|||||||
" WHERE symbol=? AND candles_json IS NOT NULL"
|
" WHERE symbol=? AND candles_json IS NOT NULL"
|
||||||
" AND logged_at >= ? ORDER BY id LIMIT ?",
|
" AND logged_at >= ? ORDER BY id LIMIT ?",
|
||||||
(f'$."{key}"', symbol, since, int(max_rows))).fetchall()
|
(f'$."{key}"', symbol, since, int(max_rows))).fetchall()
|
||||||
|
if len(rows) == int(max_rows):
|
||||||
|
logger.warning(
|
||||||
|
"signal_log: bar stream truncated for %s %s (key %s)"
|
||||||
|
" since %s — hit the %s-row cap, later bars are missing",
|
||||||
|
symbol, tf, key, since, int(max_rows))
|
||||||
for (blob,) in rows:
|
for (blob,) in rows:
|
||||||
if not blob:
|
if not blob:
|
||||||
continue
|
continue
|
||||||
@@ -1212,42 +1244,82 @@ class SignalLogger:
|
|||||||
|
|
||||||
Candidates are rows that carry indicator snapshots (a real scan)
|
Candidates are rows that carry indicator snapshots (a real scan)
|
||||||
and a price, whose signal bar closed more than 25 h ago and which
|
and a price, whose signal bar closed more than 25 h ago and which
|
||||||
have no COMPLETE label yet; the least recently labelled go first,
|
have no COMPLETE label yet. An incomplete row is re-examined only
|
||||||
so a permanently short stream can never starve the rest. Returns
|
while it is younger than SCAN_LABEL_MAX_AGE_H (counted from the
|
||||||
the number of rows actually WRITTEN — a row whose numbers did not
|
newest logged row when the database is behind the clock) — beyond
|
||||||
change is skipped, which makes repeated passes idempotent and lets
|
that its window had a real hole that no later bar can fill, so the
|
||||||
a batch driver stop at 0.
|
label it already carries is final — and a stale row never anchors
|
||||||
|
the shared bar stream either (a key that has nothing else due
|
||||||
|
starts at the age bound plus one window). Both bounds are
|
||||||
|
load-bearing: a handful of permanently incomplete rows (a feed
|
||||||
|
outage) sort to the front of every pass, and anchoring the stream
|
||||||
|
at their bar pushes its row cap past today, which silently labels
|
||||||
|
every genuinely due row NULL and incomplete — the scan_outcomes
|
||||||
|
table then goes blind. Returns the number of rows actually WRITTEN
|
||||||
|
— a row whose numbers did not change is not rewritten (only its
|
||||||
|
labeled_at is bumped, so it rotates to the back of the queue),
|
||||||
|
which makes repeated passes idempotent and lets a batch driver
|
||||||
|
stop at 0.
|
||||||
|
|
||||||
Bounded on purpose: at most ``budget`` candidates are examined and
|
Bounded on purpose: at most ``budget`` candidates are examined and
|
||||||
the pass stops after ``time_budget`` seconds (default 2 s) so the
|
the pass stops after ``time_budget`` seconds (default 2 s) so the
|
||||||
opportunistic call inside log() stays cheap. Per-row failures are
|
opportunistic call inside log() stays cheap. Per-row failures are
|
||||||
logged and skipped."""
|
logged and skipped."""
|
||||||
moment = now or datetime.now(timezone.utc)
|
moment = now or datetime.now(timezone.utc)
|
||||||
cutoff = _utc(moment) - timedelta(hours=SCAN_LABEL_MIN_AGE_H)
|
moment_utc = _utc(moment)
|
||||||
|
cutoff = moment_utc - timedelta(hours=SCAN_LABEL_MIN_AGE_H)
|
||||||
deadline = time.monotonic() + float(
|
deadline = time.monotonic() + float(
|
||||||
SCAN_LABEL_TIME_BUDGET if time_budget is None else time_budget)
|
SCAN_LABEL_TIME_BUDGET if time_budget is None else time_budget)
|
||||||
written = 0
|
written = 0
|
||||||
with self._lock:
|
with self._lock:
|
||||||
|
# Age is measured from the newest row when the database itself
|
||||||
|
# is behind the clock (an imported snapshot on the dev box),
|
||||||
|
# otherwise every row of a 3-day-old import would be declared
|
||||||
|
# final on the first pass. On a live logger the two coincide.
|
||||||
|
horizon = moment_utc
|
||||||
|
newest = self._conn.execute(
|
||||||
|
"SELECT logged_at FROM signals ORDER BY id DESC"
|
||||||
|
" LIMIT 1").fetchone()
|
||||||
|
try:
|
||||||
|
horizon = min(horizon, _utc(datetime.fromisoformat(
|
||||||
|
str(newest[0]).replace(" ", "T"))))
|
||||||
|
except (TypeError, ValueError, IndexError):
|
||||||
|
pass
|
||||||
|
stale_before = horizon - timedelta(hours=SCAN_LABEL_MAX_AGE_H)
|
||||||
|
# Deepest a stream may start when every row due for that key is
|
||||||
|
# stale: one window before the age bound, so such a row still
|
||||||
|
# gets its single final pass without dragging the stream back.
|
||||||
|
anchor_floor = stale_before - timedelta(hours=SCAN_LABEL_MIN_AGE_H)
|
||||||
|
# Row ids are monotonic in logged_at (a uniform ISO string), so
|
||||||
|
# an id bound is the cheap, index-friendly way to leave stale
|
||||||
|
# incomplete rows out of the re-examination set. bar_ts mixes
|
||||||
|
# separators and must never be compared as a string.
|
||||||
|
min_id = self._conn.execute(
|
||||||
|
"SELECT MIN(id) FROM signals WHERE logged_at >= ?",
|
||||||
|
(stale_before.isoformat(),)).fetchone()[0]
|
||||||
|
if min_id is None:
|
||||||
|
min_id = 2 ** 62 # nothing recent: re-examine none
|
||||||
rows = self._conn.execute(
|
rows = self._conn.execute(
|
||||||
"SELECT s.id, s.symbol, s.timeframe, s.price,"
|
"SELECT s.id, s.symbol, s.timeframe, s.price,"
|
||||||
" COALESCE(s.bar_ts, json_extract(s.indicators_json,"
|
" COALESCE(s.bar_ts, json_extract(s.indicators_json,"
|
||||||
" '$.\"' || s.timeframe || '\".timestamp')) AS bar,"
|
" '$.\"' || s.timeframe || '\".timestamp')) AS bar,"
|
||||||
" json_extract(s.indicators_json,"
|
" json_extract(s.indicators_json,"
|
||||||
" '$.\"' || s.timeframe || '\".key_levels.atr') AS atr"
|
" '$.\"' || s.timeframe || '\".key_levels.atr') AS atr,"
|
||||||
|
" (o.signal_id IS NULL) AS unlabelled"
|
||||||
" FROM signals s"
|
" FROM signals s"
|
||||||
" LEFT JOIN scan_outcomes o ON o.signal_id = s.id"
|
" LEFT JOIN scan_outcomes o ON o.signal_id = s.id"
|
||||||
" WHERE s.indicators_json IS NOT NULL AND s.price > 0"
|
" WHERE s.indicators_json IS NOT NULL AND s.price > 0"
|
||||||
" AND (o.signal_id IS NULL OR o.complete = 0)"
|
" AND (o.signal_id IS NULL OR (o.complete = 0 AND s.id >= ?))"
|
||||||
" AND bar IS NOT NULL"
|
" AND bar IS NOT NULL"
|
||||||
" ORDER BY COALESCE(o.labeled_at, '') ASC, s.id ASC"
|
" ORDER BY COALESCE(o.labeled_at, '') ASC, s.id ASC"
|
||||||
" LIMIT ?", (int(budget),)).fetchall()
|
" LIMIT ?", (int(min_id), int(budget))).fetchall()
|
||||||
# One bar stream per (symbol, timeframe) for the whole batch,
|
# One bar stream per (symbol, timeframe) for the whole batch,
|
||||||
# built from the OLDEST bar in it — _stream_since is a scan
|
# built from the OLDEST bar in it — _stream_since is a scan
|
||||||
# over every row logged since that bar, far too expensive to
|
# over every row logged since that bar, far too expensive to
|
||||||
# repeat per candidate.
|
# repeat per candidate.
|
||||||
due: List[Tuple[Any, ...]] = []
|
due: List[Tuple[Any, ...]] = []
|
||||||
oldest: Dict[Tuple[str, str], Tuple[datetime, str]] = {}
|
oldest: Dict[Tuple[str, str], Tuple[datetime, str]] = {}
|
||||||
for rowid, symbol, tf, price, bar, atr in rows:
|
for rowid, symbol, tf, price, bar, atr, unlabelled in rows:
|
||||||
try:
|
try:
|
||||||
stamp = _utc(datetime.fromisoformat(
|
stamp = _utc(datetime.fromisoformat(
|
||||||
str(bar).replace(" ", "T")))
|
str(bar).replace(" ", "T")))
|
||||||
@@ -1255,8 +1327,16 @@ class SignalLogger:
|
|||||||
continue
|
continue
|
||||||
if stamp > cutoff:
|
if stamp > cutoff:
|
||||||
continue # window has not elapsed yet
|
continue # window has not elapsed yet
|
||||||
|
# Stale = old by the bar it closed on AND by when it was
|
||||||
|
# written (on a live database the two agree within minutes;
|
||||||
|
# requiring both keeps a freshly imported or backfilled DB,
|
||||||
|
# whose rows all carry today's logged_at, out of it).
|
||||||
|
stale = stamp < stale_before and rowid < min_id
|
||||||
|
if stale and not unlabelled:
|
||||||
|
continue # final: a hole no later bar can ever fill
|
||||||
key = (str(symbol), str(tf))
|
key = (str(symbol), str(tf))
|
||||||
if key not in oldest or stamp < oldest[key][0]:
|
if not stale and (key not in oldest
|
||||||
|
or stamp < oldest[key][0]):
|
||||||
oldest[key] = (stamp, str(bar))
|
oldest[key] = (stamp, str(bar))
|
||||||
due.append((rowid, symbol, tf, price, bar, atr, key))
|
due.append((rowid, symbol, tf, price, bar, atr, key))
|
||||||
streams: Dict[Tuple[str, str], List[Dict[str, Any]]] = {}
|
streams: Dict[Tuple[str, str], List[Dict[str, Any]]] = {}
|
||||||
@@ -1265,8 +1345,13 @@ class SignalLogger:
|
|||||||
break
|
break
|
||||||
try:
|
try:
|
||||||
if key not in streams:
|
if key not in streams:
|
||||||
|
# 3 days of anchor is ~2,100 rows on the live DB;
|
||||||
|
# the cap is a safety net, not the working limit.
|
||||||
|
anchor = oldest.get(key)
|
||||||
streams[key] = self._stream_since(
|
streams[key] = self._stream_since(
|
||||||
symbol, tf, oldest[key][1])
|
symbol, tf,
|
||||||
|
anchor[1] if anchor else anchor_floor.isoformat(),
|
||||||
|
max_rows=10000)
|
||||||
label = scan_outcome(bar, tf, streams[key],
|
label = scan_outcome(bar, tf, streams[key],
|
||||||
_finite(atr), price)
|
_finite(atr), price)
|
||||||
if label is None:
|
if label is None:
|
||||||
@@ -1277,6 +1362,14 @@ class SignalLogger:
|
|||||||
+ " FROM scan_outcomes WHERE signal_id=?",
|
+ " FROM scan_outcomes WHERE signal_id=?",
|
||||||
(rowid,)).fetchone()
|
(rowid,)).fetchone()
|
||||||
if prior is not None and tuple(prior) == values:
|
if prior is not None and tuple(prior) == values:
|
||||||
|
# Unchanged, so NOT counted as written — but its
|
||||||
|
# queue position moves, or the same short-stream
|
||||||
|
# rows would head every pass for ever.
|
||||||
|
now_iso = datetime.now(timezone.utc).isoformat()
|
||||||
|
self._conn.execute(
|
||||||
|
"UPDATE scan_outcomes SET labeled_at=?"
|
||||||
|
" WHERE signal_id=?", (now_iso, rowid))
|
||||||
|
self._conn.commit()
|
||||||
continue
|
continue
|
||||||
self._conn.execute(
|
self._conn.execute(
|
||||||
_SCAN_OUTCOME_INSERT,
|
_SCAN_OUTCOME_INSERT,
|
||||||
@@ -1376,6 +1469,9 @@ class SignalLogger:
|
|||||||
continue
|
continue
|
||||||
before = _dumps(state)
|
before = _dumps(state)
|
||||||
advanced = advance_outcome(json.loads(before), union)
|
advanced = advance_outcome(json.loads(before), union)
|
||||||
|
if (advanced.get("status") == "UNRESOLVABLE"
|
||||||
|
and _is_gap_reason(advanced.get("reason"))):
|
||||||
|
advanced = resolve_locked_floor(advanced)
|
||||||
after = _dumps(advanced)
|
after = _dumps(advanced)
|
||||||
if after == before:
|
if after == before:
|
||||||
continue
|
continue
|
||||||
@@ -1393,7 +1489,9 @@ class SignalLogger:
|
|||||||
"""Bounded retry for continuity-broken outcomes: re-freeze the
|
"""Bounded retry for continuity-broken outcomes: re-freeze the
|
||||||
seed from the row's own setup/indicators (at most ``max_reseeds``
|
seed from the row's own setup/indicators (at most ``max_reseeds``
|
||||||
times per row) so the next resolver pass can walk the now-filled
|
times per row) so the next resolver pass can walk the now-filled
|
||||||
stream. Invalid-seed rows stay terminal."""
|
stream. Invalid-seed rows stay terminal. A gap row that had already
|
||||||
|
trailed its stop past the entry is finished at that locked floor
|
||||||
|
instead of re-seeded; the returned list holds re-seeded ids only."""
|
||||||
reseeded: List[int] = []
|
reseeded: List[int] = []
|
||||||
with self._lock:
|
with self._lock:
|
||||||
rows = self._conn.execute(
|
rows = self._conn.execute(
|
||||||
@@ -1403,8 +1501,21 @@ class SignalLogger:
|
|||||||
" AND advisory_event_id = id").fetchall()
|
" AND advisory_event_id = id").fetchall()
|
||||||
for rowid, tf, setup_json, ind_json, state_json in rows:
|
for rowid, tf, setup_json, ind_json, state_json in rows:
|
||||||
old = json.loads(state_json) if state_json else {}
|
old = json.loads(state_json) if state_json else {}
|
||||||
if "continuity" not in str(old.get("reason", "")) and \
|
if not _is_gap_reason(old.get("reason")):
|
||||||
"gap" not in str(old.get("reason", "")):
|
continue
|
||||||
|
# The locked trailing stop lives ONLY in this progress
|
||||||
|
# block, and a re-seed rebuilds the walk from scratch over
|
||||||
|
# a stream whose hole is still there (the continuity check
|
||||||
|
# is all-or-nothing), so the floor would be lost for good.
|
||||||
|
# Finish such a row here; re-seed only what has nothing.
|
||||||
|
floored = resolve_locked_floor(json.loads(_dumps(old)))
|
||||||
|
if floored.get("status") == "RESOLVED":
|
||||||
|
self._conn.execute(
|
||||||
|
"UPDATE signals SET advisory_outcome_status=?,"
|
||||||
|
" advisory_r_multiple=?, advisory_outcome_json=?"
|
||||||
|
" WHERE id=?",
|
||||||
|
("RESOLVED", canonical_r(floored),
|
||||||
|
_dumps(floored), rowid))
|
||||||
continue
|
continue
|
||||||
count = int((old.get("reseed") or {}).get("count", 0))
|
count = int((old.get("reseed") or {}).get("count", 0))
|
||||||
if count >= max_reseeds or not setup_json:
|
if count >= max_reseeds or not setup_json:
|
||||||
|
|||||||
+60
-4
@@ -49,9 +49,10 @@ import sys
|
|||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
from concurrent.futures import ThreadPoolExecutor
|
from concurrent.futures import ThreadPoolExecutor
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timedelta, timezone
|
||||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set
|
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set
|
||||||
|
|
||||||
|
import pandas as pd
|
||||||
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||||||
from telegram.constants import ParseMode
|
from telegram.constants import ParseMode
|
||||||
from telegram.error import BadRequest
|
from telegram.error import BadRequest
|
||||||
@@ -71,6 +72,7 @@ from app_config import (
|
|||||||
telegram_channel_id, violent_watch_enabled,
|
telegram_channel_id, violent_watch_enabled,
|
||||||
)
|
)
|
||||||
from providers import BaseDataProvider, make_data_provider
|
from providers import BaseDataProvider, make_data_provider
|
||||||
|
from providers.base import _is_daily_maintenance
|
||||||
from agent_panel import attach_panel, panel_health
|
from agent_panel import attach_panel, panel_health
|
||||||
from llm_client import get_llm_client
|
from llm_client import get_llm_client
|
||||||
from market_outlook import maybe_run_outlook
|
from market_outlook import maybe_run_outlook
|
||||||
@@ -110,8 +112,12 @@ _file_handler = logging.handlers.TimedRotatingFileHandler(
|
|||||||
when="midnight", backupCount=30, encoding="utf-8",
|
when="midnight", backupCount=30, encoding="utf-8",
|
||||||
)
|
)
|
||||||
_file_handler.setFormatter(_log_formatter)
|
_file_handler.setFormatter(_log_formatter)
|
||||||
|
# Console only at IMPORT: `python test_all.py` imports this module, and an
|
||||||
|
# import-time file handler wrote fixture alerts (2027 dates, 3300.00 prices)
|
||||||
|
# into logs/telegram.log on whatever box ran the suite. main() attaches
|
||||||
|
# _file_handler for real bot runs.
|
||||||
logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
|
logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
|
||||||
handlers=[_console_handler, _file_handler],
|
handlers=[_console_handler],
|
||||||
force=True)
|
force=True)
|
||||||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||||||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||||||
@@ -545,7 +551,21 @@ def _bar_epoch(tf: Timeframe, now: Optional[float] = None) -> int:
|
|||||||
|
|
||||||
|
|
||||||
def _remember_scan(symbol: str, tf: Timeframe, report: SignalReport) -> None:
|
def _remember_scan(symbol: str, tf: Timeframe, report: SignalReport) -> None:
|
||||||
_last_scan[f"{symbol}_{tf.value}"] = (_bar_epoch(tf), report)
|
# File the report under the bar it actually ANALYSED, not the wall
|
||||||
|
# clock at completion: a scan that straddles a boundary would land
|
||||||
|
# under the new bar, so the boundary loop's own scan of that bar
|
||||||
|
# found no cache entry and re-fetched/re-scored/re-logged the same
|
||||||
|
# closed bar (15 duplicate M15 rows in five days — 2026-09-13 audit).
|
||||||
|
epoch = _bar_epoch(tf)
|
||||||
|
if report.primary_analysis is not None:
|
||||||
|
try:
|
||||||
|
# report.timestamp is the primary bar's OPEN label; the bar
|
||||||
|
# closes (= the boundary this scan belongs to) one TF later.
|
||||||
|
epoch = (int(pd.Timestamp(report.timestamp).timestamp())
|
||||||
|
+ _TF_SECONDS_CACHE.get(tf, 900))
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
pass # unparseable label → wall-clock boundary, as before
|
||||||
|
_last_scan[f"{symbol}_{tf.value}"] = (epoch, report)
|
||||||
|
|
||||||
|
|
||||||
async def _cached_or_scan(
|
async def _cached_or_scan(
|
||||||
@@ -767,6 +787,16 @@ def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
|||||||
return remaining
|
return remaining
|
||||||
|
|
||||||
|
|
||||||
|
def _closed_bar_in_settlement(interval_minutes: int,
|
||||||
|
now: Optional[datetime] = None) -> bool:
|
||||||
|
"""True when the bar that just closed (label = now - interval) falls in
|
||||||
|
gold's daily settlement pause — no such bar exists, so a scan would only
|
||||||
|
re-read the previous bar (61 stale rows in the 2026-09 audit DB)."""
|
||||||
|
label_time = ((now or datetime.now(timezone.utc))
|
||||||
|
- timedelta(minutes=interval_minutes))
|
||||||
|
return _is_daily_maintenance(label_time)
|
||||||
|
|
||||||
|
|
||||||
# ── Violent-move watch (display-only) ────────────────────────────────
|
# ── Violent-move watch (display-only) ────────────────────────────────
|
||||||
# Per-process heads-up budget: {"day": date, "count": int, "sent": {...}}.
|
# Per-process heads-up budget: {"day": date, "count": int, "sent": {...}}.
|
||||||
_watch_state: Dict[str, Any] = {}
|
_watch_state: Dict[str, Any] = {}
|
||||||
@@ -800,7 +830,11 @@ async def _scan_timeframe(
|
|||||||
confirmation_tfs: List[Timeframe],
|
confirmation_tfs: List[Timeframe],
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Scan all pairs for one timeframe and broadcast results."""
|
"""Scan all pairs for one timeframe and broadcast results."""
|
||||||
tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS]
|
# _cached_or_scan, not run_signal: the WS breach handler may already
|
||||||
|
# have scanned this closed bar (and vice versa) — one fetch/score/log
|
||||||
|
# per bar per TF (2026-09-13 audit).
|
||||||
|
tasks = [_cached_or_scan(symbol, tf, confirmation_tfs)
|
||||||
|
for symbol in SCAN_PAIRS]
|
||||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||||
reports: List[SignalReport] = []
|
reports: List[SignalReport] = []
|
||||||
for i, result in enumerate(results):
|
for i, result in enumerate(results):
|
||||||
@@ -930,6 +964,9 @@ _pending_breach: Dict[str, tuple] = {}
|
|||||||
# A failed breach rescan shrinks the just-set cooldown to this retry
|
# A failed breach rescan shrinks the just-set cooldown to this retry
|
||||||
# window instead of silencing the symbol for the full cooldown (H33a).
|
# window instead of silencing the symbol for the full cooldown (H33a).
|
||||||
_BREACH_RETRY_SECONDS = 30
|
_BREACH_RETRY_SECONDS = 30
|
||||||
|
# A breach landing inside this many seconds after an M15 boundary waits for
|
||||||
|
# the candle loop's scan (boundary+10s) and reuses it — 2026-09-13 audit.
|
||||||
|
_BREACH_DEFER_SECONDS = 20.0
|
||||||
|
|
||||||
_cfg = PlatformConfig()
|
_cfg = PlatformConfig()
|
||||||
_WS_COOLDOWN = _cfg.ws_cooldown_seconds
|
_WS_COOLDOWN = _cfg.ws_cooldown_seconds
|
||||||
@@ -1003,6 +1040,13 @@ async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) -
|
|||||||
logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15")
|
logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
# A tick a second after an M15 boundary would scan a bar the feed
|
||||||
|
# may still be lagging on, and the candle loop (boundary+10s) would
|
||||||
|
# then scan the same bar again. Wait for the loop and reuse it.
|
||||||
|
elapsed = time.time() - _bar_epoch(Timeframe.M15)
|
||||||
|
if elapsed < _BREACH_DEFER_SECONDS:
|
||||||
|
logger.debug("deferring breach rescan to the boundary scan")
|
||||||
|
await asyncio.sleep(_BREACH_DEFER_SECONDS - elapsed)
|
||||||
report = await _cached_or_scan(pair_symbol, Timeframe.M15)
|
report = await _cached_or_scan(pair_symbol, Timeframe.M15)
|
||||||
_update_level_cache([report])
|
_update_level_cache([report])
|
||||||
except Exception:
|
except Exception:
|
||||||
@@ -1152,6 +1196,11 @@ async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
|
|||||||
await asyncio.sleep(wait)
|
await asyncio.sleep(wait)
|
||||||
if not _feed_active():
|
if not _feed_active():
|
||||||
break
|
break
|
||||||
|
if _closed_bar_in_settlement(interval_minutes):
|
||||||
|
logger.info(
|
||||||
|
f"{label} auto-scan: skipped — bar label inside the "
|
||||||
|
f"daily settlement pause")
|
||||||
|
continue
|
||||||
logger.info(f"{label} auto-scan: firing")
|
logger.info(f"{label} auto-scan: firing")
|
||||||
try:
|
try:
|
||||||
await _scan_timeframe(
|
await _scan_timeframe(
|
||||||
@@ -1824,6 +1873,13 @@ def main() -> None:
|
|||||||
"and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and "
|
"and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and "
|
||||||
"TELEGRAM_CHAT_ID environment variables.")
|
"TELEGRAM_CHAT_ID environment variables.")
|
||||||
|
|
||||||
|
# Only a real bot run writes the rotating log file: tests import this
|
||||||
|
# module (an import-time handler logged their fixtures into
|
||||||
|
# logs/telegram.log) and one of them calls main() to assert the refusal
|
||||||
|
# above, so the attach sits after that guard — 2026-09-13 audit.
|
||||||
|
if _file_handler not in logging.getLogger().handlers:
|
||||||
|
logging.getLogger().addHandler(_file_handler)
|
||||||
|
|
||||||
_install_excepthooks()
|
_install_excepthooks()
|
||||||
logger.info("Starting ForexBot2 Telegram bot…")
|
logger.info("Starting ForexBot2 Telegram bot…")
|
||||||
|
|
||||||
|
|||||||
+1015
-6
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user