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

audit: alert delivery, provider robustness, backtester NaN/cache guards (Batches C+E+F)

Browse Source 
2026-08-04 Fable 5 QA audit. Three batches sharing one test file, all
verified by runnable repros before fixing, all gated below.

Batch C — telegram alert delivery (verified by h4 harness):
- Failed breach rescan shrinks the cooldown to 30 s instead of
  silencing the symbol for 5 min; breach during cooldown is PARKED and
  fired on the first tick after expiry (was dropped forever).
- Level cache keyed symbol→TF (M5 scans no longer wipe H1/H4 swing
  levels within minutes); breach labels TF-qualified.
- Markdown alert sends get a plain-text fallback (mirroring _safe_edit);
  fingerprint commits only when a message actually landed.
- Per-key asyncio.Lock closes the dedup TOCTOU between auto-scan and
  breach sends (forced-interleave repro produced duplicate alerts).
- _start_scheduler respawns dead per-TF loops (was all-or-nothing).
- Clean WS close now backs off (was a zero-delay reconnect hot loop);
  backoff resets only after a connection that received ticks; TD
  iter_ticks fires on_connect on the first RECEIVED frame (invalid key
  can no longer pin the backoff at minimum).
- Scan-failure logs carry the real traceback (was "NoneType: None").

Batch E — providers (verified by h5 harness):
- fetch_range: 3 consecutive chunk ERRORS raise instead of returning
  silently truncated history (verified 731-day silent gap repro; also
  covers 429 storms); pagination no-progress guard; head-truncation
  ERROR log.
- Single-flight per (symbol,TF): 4 concurrent scans = 1 API call (was
  4); cache records requested depth so a longer-lookback call refetches.
- Multi-bar forming/future tail fully dropped (was iloc[:-1] once).
- Oanda: malformed candles skipped, never 0.0-defaulted; cleaning drops
  non-positive-price rows BEFORE OHLC repair; inf returns neutralized so
  one corrupt bar can no longer disarm the whole 10-sigma filter.

Batch F — backtester internals (verified by h6 harness):
- NaN bar in a trailing window exits at the last valid bar (was NaN R
  poisoning every metric); static-path TIMEOUT NaN guard;
  compute_metrics excludes non-finite R loudly. Zero real-data exposure
  (0 NaN rows in all 10 Oanda parquets) — defensive.
- fetch_history ERROR-logs pre-2026-07-08 cache files (cleaned-bars era)
  when loaded for raw-path resolution.

Gates: suite 333/333 green; pinned M15 12mo book reproduced identically
(397 trades, +78.7111R) after E and after F; M5 HEAD-config baseline
reproduced identically (556 trades, +97.5604R) after F.

Claude-Session: https://claude.ai/code/session_01UnZkzeFcTu9agFg8BDh2tn
This commit is contained in:
Harwinder Sekhon
2026-08-04 20:02:57 -04:00
parent 18e4fd0403
commit 99eb268cf3
6 changed files with 1117 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
providers/base.py
+153 -62
View File
@@ -86,9 +86,15 @@ class BaseDataProvider(ABC):
self.config = config
self._cache: Dict[str, pd.DataFrame] = {}
self._cache_timestamps: Dict[str, float] = {}
# Requested fetch depth per key: a cached short frame must not
# satisfy a caller asking for a longer lookback.
self._cache_outputsize: Dict[str, int] = {}
# The bot shares one provider across a thread pool; cache eviction
# is check-then-act and needs locking.
self._cache_lock = threading.Lock()
# Single-flight: one API fetch per (symbol, TF) at a time; other
# threads wait and then read the cache.
self._inflight: Dict[str, threading.Event] = {}
# ── Abstract: the only provider-specific piece ─────────────────
@@ -119,68 +125,97 @@ class BaseDataProvider(ABC):
lookback_days = self.config.data_lookback_days
cache_key = f"{symbol}_{timeframe.value}"
now = time.time()
cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval)
with self._cache_lock:
cached = self._cache.get(cache_key)
cached_at = self._cache_timestamps.get(cache_key)
if (
cached is not None
and cached_at is not None
and (now - cached_at) < cache_ttl
and not cached.empty
):
if (self._is_data_fresh(cached, symbol, timeframe)
and self._has_latest_closed_bar(cached, timeframe)):
return cached
logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
with self._cache_lock:
self._cache.pop(cache_key, None)
self._cache_timestamps.pop(cache_key, None)
outputsize = max(
300,
min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20,
self.MAX_BARS_PER_REQUEST),
)
df = self._fetch_raw(symbol, timeframe, outputsize)
df = self._drop_forming_candle(df, timeframe)
df = self._validate_and_clean(df, symbol, timeframe)
# The closed-bar guarantee must hold for FRESH responses too, not
# just cache hits: the scanner fires seconds after the boundary,
# and a lagging feed can answer without the just-closed bar — the
# scan would silently analyze (and session-check) the previous
# bar. Give the feed one short grace retry, then proceed loudly.
if not df.empty and not self._has_latest_closed_bar(df, timeframe):
time.sleep(self.FRESH_FETCH_RETRY_SECONDS)
try:
retry = self._fetch_raw(symbol, timeframe, outputsize)
retry = self._drop_forming_candle(retry, timeframe)
retry = self._validate_and_clean(retry, symbol, timeframe)
if retry is not None and not retry.empty:
df = retry
except DataIntegrityError as e:
while True:
now = time.time()
cache_ttl = self.CANDLE_SECONDS.get(
timeframe, self.config.refresh_interval)
with self._cache_lock:
cached = self._cache.get(cache_key)
cached_at = self._cache_timestamps.get(cache_key)
cached_size = self._cache_outputsize.get(cache_key, 0)
if (
cached is not None
and cached_at is not None
and (now - cached_at) < cache_ttl
and not cached.empty
# A caller asking for MORE history than the cached fetch
# requested must not be served the short frame silently.
and cached_size >= outputsize
):
if (self._is_data_fresh(cached, symbol, timeframe)
and self._has_latest_closed_bar(cached, timeframe)):
return cached
logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
with self._cache_lock:
self._cache.pop(cache_key, None)
self._cache_timestamps.pop(cache_key, None)
self._cache_outputsize.pop(cache_key, None)
# Single-flight: concurrent scans for the same (symbol, TF)
# must not each hit the API (the four TF loops fire together
# at aligned boundaries and Oanda has no client throttle).
# First caller claims the key and fetches; the rest wait,
# then re-check the cache.
with self._cache_lock:
inflight = self._inflight.get(cache_key)
if inflight is None:
self._inflight[cache_key] = threading.Event()
break
inflight.wait(timeout=90)
try:
df = self._fetch_raw(symbol, timeframe, outputsize)
df = self._drop_forming_candle(df, timeframe)
df = self._validate_and_clean(df, symbol, timeframe)
# The closed-bar guarantee must hold for FRESH responses too,
# not just cache hits: the scanner fires seconds after the
# boundary, and a lagging feed can answer without the
# just-closed bar — the scan would silently analyze (and
# session-check) the previous bar. Give the feed one short
# grace retry, then proceed loudly.
if not df.empty and not self._has_latest_closed_bar(df, timeframe):
time.sleep(self.FRESH_FETCH_RETRY_SECONDS)
try:
retry = self._fetch_raw(symbol, timeframe, outputsize)
retry = self._drop_forming_candle(retry, timeframe)
retry = self._validate_and_clean(retry, symbol, timeframe)
if retry is not None and not retry.empty:
df = retry
except DataIntegrityError as e:
logger.warning(
f"{symbol} {timeframe.value}: boundary retry refetch "
f"failed ({e}) — keeping original data")
if not self._has_latest_closed_bar(df, timeframe):
logger.warning(
f"{symbol} {timeframe.value}: feed lagging — just-closed bar "
f"still missing after retry; this scan may run one bar behind"
)
if not self._is_data_fresh(df, symbol, timeframe):
logger.warning(
f"{symbol} {timeframe.value}: boundary retry refetch "
f"failed ({e}) — keeping original data")
if not self._has_latest_closed_bar(df, timeframe):
logger.warning(
f"{symbol} {timeframe.value}: feed lagging — just-closed bar "
f"still missing after retry; this scan may run one bar behind"
f"STALE DATA from API: {symbol} {timeframe.value} — "
f"last bar {df.index[-1]} is too old. Signal quality may be degraded."
)
if not self._is_data_fresh(df, symbol, timeframe):
logger.warning(
f"STALE DATA from API: {symbol} {timeframe.value} — "
f"last bar {df.index[-1]} is too old. Signal quality may be degraded."
)
with self._cache_lock:
self._cache[cache_key] = df
self._cache_timestamps[cache_key] = now
return df
with self._cache_lock:
self._cache[cache_key] = df
self._cache_timestamps[cache_key] = now
self._cache_outputsize[cache_key] = outputsize
return df
finally:
# Release waiters even when the fetch raised — they re-check
# the cache and (on a miss) claim the key themselves.
with self._cache_lock:
ev = self._inflight.pop(cache_key, None)
if ev is not None:
ev.set()
def fetch_range(
self,
@@ -199,6 +234,7 @@ class BaseDataProvider(ABC):
chunks = []
cursor_end = end
empty_streak = 0
error_streak = 0
while cursor_end > start:
cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
try:
@@ -207,19 +243,27 @@ class BaseDataProvider(ABC):
start=cursor_start, end=cursor_end,
)
except Exception as e:
# Providers raise for windows with no data (weekends,
# holidays, asset gaps). Skip the window, step back, retry
# — but bail after a few consecutive empties.
# A raising chunk is a PROVIDER ERROR (auth, 429, 5xx,
# malformed payload) — not "no data". Three in a row means
# the rest of the history is unreachable: raise instead of
# returning a silently truncated frame (the backtester
# would treat it as complete).
logger.warning(
f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
)
empty_streak += 1
if empty_streak >= 3:
break
error_streak += 1
if error_streak >= 3:
raise self.DATA_ERROR(
f"{symbol} {timeframe.value}: {error_streak} "
f"consecutive chunk failures walking back from "
f"{cursor_end.date()} — refusing to return "
f"truncated history (last error: {e})") from e
cursor_end = cursor_start - timedelta(seconds=1)
continue
if chunk is None or chunk.empty:
# Genuinely empty windows (weekends/holidays/asset gaps)
# are normal; several in a row = start of history.
empty_streak += 1
if empty_streak >= 3:
break
@@ -227,9 +271,18 @@ class BaseDataProvider(ABC):
continue
empty_streak = 0
error_streak = 0
chunks.append(chunk)
# Step back; dedup on concat.
cursor_end = chunk.index[0] - timedelta(seconds=1)
new_cursor = chunk.index[0] - timedelta(seconds=1)
if new_cursor >= cursor_end:
# A provider ignoring start/end would pin the cursor and
# loop forever — refuse to spin.
logger.warning(
f"{symbol} {timeframe.value}: pagination made no "
f"progress at {cursor_end} — stopping")
break
cursor_end = new_cursor
if len(chunk) < 2:
break
@@ -243,7 +296,17 @@ class BaseDataProvider(ABC):
df = self._drop_forming_candle(df, timeframe)
start_ts = pd.Timestamp(start)
end_ts = pd.Timestamp(end)
return df[(df.index >= start_ts) & (df.index <= end_ts)]
df = df[(df.index >= start_ts) & (df.index <= end_ts)]
# Coverage sanity: a head materially later than the requested
# start means history was truncated (empty-streak bail or provider
# depth limit). Loud, not fatal — some assets genuinely start late.
if not df.empty and (df.index[0] - start_ts) > pd.Timedelta(days=7):
logger.error(
f"{symbol} {timeframe.value}: fetched history starts "
f"{df.index[0]} but {start_ts} was requested — "
f"~{(df.index[0] - start_ts).days} days missing from the "
f"head; treat downstream results accordingly")
return df
def is_data_fresh(
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
@@ -323,8 +386,19 @@ class BaseDataProvider(ABC):
return df
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
now = pd.Timestamp.now(tz="UTC")
if df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now:
# Loop, not a single slice: a misbehaving feed can return MORE
# than one not-yet-closed bar, and a single iloc[:-1] would let
# the residual forming bar through cleaning (future-tolerance is
# +5 min) straight into the signal engine.
dropped = 0
while (not df.empty
and df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now):
df = df.iloc[:-1]
dropped += 1
if dropped > 1:
logger.warning(
"%s: feed returned %d forming/future bars — all dropped",
timeframe.value, dropped)
return df
def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool:
@@ -372,6 +446,20 @@ class BaseDataProvider(ABC):
logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps")
df = df[~future_mask]
# Non-positive prices are corrupt, never real gold quotes. Reject
# the row BEFORE the consistency repair below — repairing High/Low
# from a corrupt 0.0 Open/Close would propagate the zero into
# previously-valid fields (2026-08-04 audit, I45).
nonpositive = (df[["Open", "High", "Low", "Close"]] <= 0).any(axis=1)
if nonpositive.any():
logger.warning(
f"{symbol}: Dropping {nonpositive.sum()} bars with "
f"non-positive prices (corrupt feed data)")
df = df[~nonpositive]
if df.empty:
raise self.DATA_ERROR(
f"{symbol}: all bars had non-positive prices")
invalid_ohlc = (
(df["High"] < df["Low"]) |
(df["High"] < df["Open"]) |
@@ -386,7 +474,10 @@ class BaseDataProvider(ABC):
df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]])
df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]])
# inf returns (division by a zero/na Close) would NaN the std and
# silently disarm the whole 10σ pass — neutralize them first.
returns = df["Close"].pct_change()
returns = returns.replace([float("inf"), float("-inf")], float("nan"))
std_ret = returns.std()
if std_ret > 0:
dev = returns - returns.mean()