Harwinder Sekhon
7dbcbc6c82
Owner decisions from the 2026-08-04 audit report: - spread-in-exit-triggers: DECLINED (books stay comparable; ledger in CLAUDE.md records the measured ~4-6% optimism). - weekend clock: FIXED. The market week is Fri 17:00 -> Sun 18:00 New York time, not Fri/Sun 22:00 UTC — during US DST the Oanda feed stops at 21:00 UTC (measured: 159/162 summer weekends). All three weekend implementations (providers/base, signal_aggregator, gpu_backtest calendar filter) now share the NY-anchored rule; a new test sweeps them hourly across both DST transitions and asserts agreement. No NTP needed — the host clock was fine, the constant was wrong. Ends the weekly false-"STALE DATA / feed lagging" hours (summer Fri 21-22 UTC, winter Sun 22-23 UTC). - FB2_EMA_LO (delegated): kept as a research knob but crash-proofed — an unparseable value degrades to the tuned default 0.5 with a warning instead of killing every scan. Documented in CLAUDE.md. Book gates: M15 12mo pinned book byte-identical (+78.7111R, 397 trades). M5 12mo drops exactly ONE trade (556 -> 555, +97.56 -> +98.64R): a phantom whose signal fired on the final summer-Friday bar (decision 21:00 UTC = NY 17:00 market close) and was "filled" at Sunday's reopen 49 hours later, -1.08R — an entry live trading can never take. The backtest no longer contains it; live == backtest parity improved. M15 unaffected because hour 21 is in its blocked set. Suite: 341/341 green. Claude-Session: https://claude.ai/code/session_01UnZkzeFcTu9agFg8BDh2tn
533 lines
22 KiB
Python
533 lines
22 KiB
Python
"""
|
||
ForexBot2 — Provider base class
|
||
================================
|
||
`BaseDataProvider` owns everything that must behave identically no matter
|
||
where the bars come from: in-memory caching with candle-duration TTL, the
|
||
closed-bar boundary guarantee, freshness checks, validation/cleaning, the
|
||
forming-candle drop, and paginated date-range fetching for the backtester.
|
||
|
||
A concrete provider implements only:
|
||
_fetch_raw(symbol, timeframe, outputsize, start, end) -> pd.DataFrame
|
||
returning a normalized frame (capitalized OHLCV columns, UTC tz-aware
|
||
DatetimeIndex named "Datetime", oldest-first), plus optionally
|
||
`get_spread` and the streaming-quote capability (`supports_streaming` +
|
||
`iter_ticks`).
|
||
|
||
The signal pipeline (engine + aggregator + backtester) consumes ONLY the
|
||
public contract defined here, so swapping providers never touches signal
|
||
code.
|
||
"""
|
||
|
||
import logging
|
||
import threading
|
||
import time
|
||
from abc import ABC, abstractmethod
|
||
from datetime import datetime, timedelta
|
||
from typing import AsyncIterator, Callable, Dict, Optional, Sequence, Tuple
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import pandas as pd
|
||
|
||
from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# The gold/FX market week is defined in New York time (see
|
||
# _is_forex_weekend); a fixed UTC hour is wrong for half the year.
|
||
_NY_TZ = ZoneInfo("America/New_York")
|
||
|
||
|
||
def _is_forex_weekend(now=None) -> bool:
|
||
"""Gold/FX market closed for the weekend.
|
||
|
||
The market week is anchored to NEW YORK time — Friday 17:00 ET →
|
||
Sunday 18:00 ET — not a fixed UTC hour. During US DST the feed
|
||
actually stops at 21:00 UTC (measured on 5 years of Oanda bars:
|
||
159/162 summer Fridays; winter close 22:00, winter reopen Sunday
|
||
23:00 UTC). The old hardcoded Fri 22:00→Sun 22:00 UTC rule produced
|
||
two false-"STALE DATA/feed lagging" hours every week (2026-08-04
|
||
audit I43, fixed per owner decision).
|
||
|
||
Kept as three deliberate copies (here, signal_aggregator.py,
|
||
gpu_backtest._is_xau_blocked_by_calendar) — a Module 21 test asserts
|
||
they agree hour-by-hour across DST transitions.
|
||
"""
|
||
if now is None:
|
||
now = pd.Timestamp.now(tz="UTC")
|
||
ny = now.astimezone(_NY_TZ)
|
||
wd = ny.weekday()
|
||
if wd == 4: # Friday
|
||
return ny.hour >= 17
|
||
if wd == 5: # Saturday
|
||
return True
|
||
if wd == 6: # Sunday
|
||
return ny.hour < 18
|
||
return False
|
||
|
||
|
||
class DataIntegrityError(Exception):
|
||
"""Raised when data fails validation checks."""
|
||
pass
|
||
|
||
|
||
class BaseDataProvider(ABC):
|
||
"""Shared OHLCV provider behavior; subclasses supply the raw fetch."""
|
||
|
||
#: Registry name; overridden per provider.
|
||
name: str = "base"
|
||
|
||
#: Exception type this provider raises for fetch/validation failures.
|
||
#: Provider-specific errors subclass DataIntegrityError so consumers
|
||
#: can catch one type regardless of the active provider.
|
||
DATA_ERROR = DataIntegrityError
|
||
|
||
BARS_PER_DAY = {
|
||
Timeframe.M1: 1440, Timeframe.M5: 288, Timeframe.M15: 96,
|
||
Timeframe.M30: 48, Timeframe.H1: 24, Timeframe.H4: 6,
|
||
Timeframe.D1: 1, Timeframe.W1: 0.143,
|
||
}
|
||
|
||
CANDLE_SECONDS = {
|
||
Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900,
|
||
Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400,
|
||
Timeframe.D1: 86400, Timeframe.W1: 604800,
|
||
}
|
||
|
||
#: Hard per-request bar cap shared by Twelve Data and Oanda.
|
||
MAX_BARS_PER_REQUEST = 5000
|
||
|
||
# Grace wait before the one retry when a fresh response at a candle
|
||
# boundary is missing the just-closed bar (feed latency).
|
||
FRESH_FETCH_RETRY_SECONDS = 3.0
|
||
|
||
def __init__(self, config: PlatformConfig):
|
||
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 ─────────────────
|
||
|
||
@abstractmethod
|
||
def _fetch_raw(
|
||
self,
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
outputsize: int,
|
||
start: Optional[datetime] = None,
|
||
end: Optional[datetime] = None,
|
||
) -> pd.DataFrame:
|
||
"""Fetch up to `outputsize` bars, optionally bounded by start/end
|
||
(tz-aware UTC datetimes). Must return a normalized frame:
|
||
capitalized OHLCV columns, UTC DatetimeIndex named "Datetime",
|
||
oldest-first. Raise self.DATA_ERROR on failure."""
|
||
|
||
# ── Public API ─────────────────────────────────────────────────
|
||
|
||
def fetch_ohlcv(
|
||
self,
|
||
symbol: str = XAUUSD,
|
||
timeframe: Timeframe = Timeframe.H1,
|
||
lookback_days: Optional[int] = None,
|
||
) -> pd.DataFrame:
|
||
"""Fetch OHLCV with in-memory caching for live mode."""
|
||
if lookback_days is None:
|
||
lookback_days = self.config.data_lookback_days
|
||
|
||
cache_key = f"{symbol}_{timeframe.value}"
|
||
outputsize = max(
|
||
300,
|
||
min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20,
|
||
self.MAX_BARS_PER_REQUEST),
|
||
)
|
||
|
||
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"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
|
||
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,
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
start: datetime,
|
||
end: datetime,
|
||
) -> pd.DataFrame:
|
||
"""Fetch the RAW closed-bar series between start/end, paginating
|
||
under the per-request bar cap. Returns uncleaned bars (the
|
||
backtester resolves trades on the raw price path); callers wanting
|
||
the signal-side series run `validate_and_clean` on the result."""
|
||
bars_per_day = self.BARS_PER_DAY[timeframe]
|
||
days_per_chunk = max(1, int(4500 / bars_per_day))
|
||
|
||
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:
|
||
chunk = self._fetch_raw(
|
||
symbol, timeframe, self.MAX_BARS_PER_REQUEST,
|
||
start=cursor_start, end=cursor_end,
|
||
)
|
||
except Exception as e:
|
||
# 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}"
|
||
)
|
||
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
|
||
cursor_end = cursor_start - timedelta(seconds=1)
|
||
continue
|
||
|
||
empty_streak = 0
|
||
error_streak = 0
|
||
chunks.append(chunk)
|
||
# Step back; dedup on concat.
|
||
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
|
||
|
||
if not chunks:
|
||
raise RuntimeError(f"No data fetched for {symbol} {timeframe.value}")
|
||
|
||
df = pd.concat(chunks).sort_index()
|
||
df = df[~df.index.duplicated(keep="last")]
|
||
# Fetches ending "now" include the still-forming candle — drop it
|
||
# so callers only ever see closed bars.
|
||
df = self._drop_forming_candle(df, timeframe)
|
||
start_ts = pd.Timestamp(start)
|
||
end_ts = pd.Timestamp(end)
|
||
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
|
||
) -> bool:
|
||
"""Public freshness check (last bar within 3 candle durations)."""
|
||
return self._is_data_fresh(df, symbol, timeframe)
|
||
|
||
def validate_and_clean(
|
||
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
|
||
) -> pd.DataFrame:
|
||
"""Public entry to the shared cleaning pipeline."""
|
||
return self._validate_and_clean(df, symbol, timeframe)
|
||
|
||
def clear_cache(self) -> int:
|
||
"""Wipe all cached data. Returns number of entries cleared."""
|
||
with self._cache_lock:
|
||
count = len(self._cache)
|
||
self._cache.clear()
|
||
self._cache_timestamps.clear()
|
||
logger.info(f"Cache cleared: {count} entries wiped")
|
||
return count
|
||
|
||
def get_pair_info(self, symbol: str = XAUUSD) -> Dict:
|
||
info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])
|
||
return {"symbol": symbol, **info}
|
||
|
||
def get_spread(self, symbol: str = XAUUSD) -> float:
|
||
"""Latest observed round-trip spread; providers without live
|
||
bid/ask fall back to the pair's assumed default."""
|
||
return PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"]
|
||
|
||
# ── Streaming-quote capability (optional) ──────────────────────
|
||
|
||
#: True when the provider can stream live ticks via `iter_ticks`.
|
||
supports_streaming: bool = False
|
||
|
||
def iter_ticks(
|
||
self,
|
||
symbols: Sequence[str],
|
||
on_connect: Optional[Callable[[], None]] = None,
|
||
) -> AsyncIterator[Tuple[str, float]]:
|
||
"""Async generator yielding (canonical_symbol, price) ticks for one
|
||
connection's lifetime; the caller owns reconnect/backoff. Providers
|
||
that support it set `supports_streaming = True` and override."""
|
||
raise NotImplementedError(
|
||
f"{self.name} provider does not support streaming quotes")
|
||
|
||
# ── Shared internals ───────────────────────────────────────────
|
||
|
||
def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool:
|
||
"""Cached data must include the most recent fully closed bar.
|
||
|
||
The TTL alone is wall-clock based: a snapshot fetched mid-candle is
|
||
still within TTL at the next candle boundary but is missing the bar
|
||
that just closed, so every boundary scan would analyze — and run the
|
||
bar-close session check against — the previous bar.
|
||
"""
|
||
if df is None or df.empty:
|
||
return False
|
||
now = pd.Timestamp.now(tz="UTC")
|
||
if _is_forex_weekend(now):
|
||
return True
|
||
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
|
||
dur = pd.Timedelta(seconds=candle_seconds)
|
||
expected_label = now.floor(dur) - dur
|
||
return df.index[-1] >= expected_label
|
||
|
||
def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame:
|
||
"""Drop the still-forming last candle so analysis only sees closed bars.
|
||
|
||
Some feeds include the current partial candle; the scanner fires
|
||
seconds after the boundary, so without this guard every indicator
|
||
reads a seconds-old stub bar and signals repaint. Feeds that only
|
||
return complete candles pass through unchanged.
|
||
"""
|
||
if df is None or df.empty:
|
||
return df
|
||
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
|
||
now = pd.Timestamp.now(tz="UTC")
|
||
# 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:
|
||
"""Check if the last candle is within acceptable age for this timeframe."""
|
||
if df is None or df.empty:
|
||
return False
|
||
|
||
now = pd.Timestamp.now(tz="UTC")
|
||
|
||
if _is_forex_weekend(now):
|
||
return True
|
||
|
||
last_bar = df.index[-1]
|
||
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
|
||
max_age_seconds = candle_seconds * 3
|
||
age_seconds = (now - last_bar).total_seconds()
|
||
|
||
if age_seconds > max_age_seconds:
|
||
logger.warning(
|
||
f"{symbol} {timeframe.value}: last bar {last_bar.strftime('%Y-%m-%d %H:%M')} "
|
||
f"is {age_seconds / 60:.0f}m old (limit: {max_age_seconds / 60:.0f}m)"
|
||
)
|
||
return False
|
||
|
||
return True
|
||
|
||
def _validate_and_clean(
|
||
self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
|
||
) -> pd.DataFrame:
|
||
if df is None or df.empty:
|
||
raise self.DATA_ERROR(f"Empty dataset for {symbol}")
|
||
|
||
original_len = len(df)
|
||
|
||
df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all")
|
||
df = df.ffill(limit=3)
|
||
df = df.dropna(subset=["Close"])
|
||
|
||
df = df.sort_index()
|
||
df = df[~df.index.duplicated(keep="last")]
|
||
|
||
now = pd.Timestamp.utcnow()
|
||
future_mask = df.index > now + pd.Timedelta(minutes=5)
|
||
if future_mask.any():
|
||
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"]) |
|
||
(df["High"] < df["Close"]) |
|
||
(df["Low"] > df["Open"]) |
|
||
(df["Low"] > df["Close"])
|
||
)
|
||
if invalid_ohlc.any():
|
||
logger.warning(f"{symbol}: Fixing {invalid_ohlc.sum()} OHLC-inconsistent bars")
|
||
for idx in df[invalid_ohlc].index:
|
||
vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]]
|
||
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()
|
||
outlier_mask = dev.abs() > 10 * std_ret
|
||
# A spike bar makes the NEXT bar's return look extreme too (it is
|
||
# measured against the spike). If the previous bar is flagged and
|
||
# this bar reverses direction, it is the legitimate snap-back —
|
||
# keep it, or every spike would punch a two-bar hole in the data.
|
||
recovery = (
|
||
outlier_mask
|
||
& outlier_mask.shift(1, fill_value=False)
|
||
& (dev * dev.shift(1) < 0)
|
||
)
|
||
outlier_mask = outlier_mask & ~recovery
|
||
if outlier_mask.any():
|
||
logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars")
|
||
df = df[~outlier_mask]
|
||
|
||
min_required = max(
|
||
self.config.technical.ema_trend,
|
||
200,
|
||
) + 10
|
||
|
||
if len(df) < min_required:
|
||
logger.warning(
|
||
f"{symbol} {timeframe.value}: Only {len(df)} bars "
|
||
f"(need {min_required}). Some indicators will be partial."
|
||
)
|
||
|
||
cleaned = original_len - len(df)
|
||
if cleaned > 0:
|
||
logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars")
|
||
|
||
return df
|