Compare commits
7 Commits
3c9bb2f017
...
fc5ce85493
| Author | SHA1 | Date | |
|---|---|---|---|
 Harwinder Sekhon
|
fc5ce85493 | ||
|
Harwinder Sekhon
|
45ce97ed13 | ||
|
Harwinder Sekhon
|
8eca2f4044 | ||
|
Harwinder Sekhon
|
fc40ba081f | ||
|
Harwinder Sekhon
|
f7924af3b0 | ||
|
Harwinder Sekhon
|
7395aa5473 | ||
|
Harwinder Sekhon
|
c9829a2a67 |
@@ -11,7 +11,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest
|
||||
## Commands
|
||||
|
||||
```bash
|
||||
python test_all.py # QA suite: 302 tests, 20 modules, no network (mocked)
|
||||
python test_all.py # QA suite: 338 tests, 21 modules, no network (mocked)
|
||||
python main.py --pair XAUUSD --timeframe H1 --detail # CLI signal report (--json for full dump)
|
||||
python telegram_bot.py # the bot (refuses to boot without credentials)
|
||||
python oanda_daemon.py # auto-execution daemon on Oanda practice account
|
||||
@@ -26,7 +26,7 @@ python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-0
|
||||
# --warmup-bars --cooldown-bars --no-cache --export-csv out.csv --detail
|
||||
```
|
||||
|
||||
**`test_all.py` is NOT pytest** — it uses a custom `@test("name")` decorator that runs each test immediately at import; don't grep for `def test_` or try `pytest test_all.py`. Run the whole file; it prints per-module results and exits non-zero on failure. There is no way to run a single test short of commenting modules out. Tests must stay green; every new gate/knob needs tests. When patching weekend logic in tests, patch `providers.base._is_forex_weekend`.
|
||||
**`test_all.py` is NOT pytest** — it uses a custom `@test("name")` decorator that runs each test immediately at import; don't grep for `def test_` or try `pytest test_all.py`. Run the whole file; it prints per-module results and exits non-zero on failure. There is no way to run a single test short of commenting modules out. Tests must stay green; every new gate/knob needs tests. When patching weekend logic in tests, patch `providers.base._is_forex_weekend` — but know there are THREE independent weekend implementations (`providers/base.py`, `signal_aggregator.py`, and an inline copy in `backtest/gpu_backtest.py:_is_xau_blocked_by_calendar`); patching the provider one does not reach the other two.
|
||||
|
||||
Python 3.12; a `venv/` exists in-repo. Dependencies in `requirements.txt` (numpy, pandas, requests, twelvedata, python-telegram-bot ≥20, websockets).
|
||||
|
||||
@@ -39,7 +39,8 @@ Precedence for every credential/deployment setting: **env var → config.json
|
||||
### End-to-end pipeline (every scan)
|
||||
|
||||
```
|
||||
providers/<active>.fetch_ohlcv() per TF: throttle, fetch, drop forming candle, validate/clean, cache
|
||||
providers/<active>.fetch_ohlcv() per TF: single-flight fetch, drop forming candle, validate/clean, cache
|
||||
(only Twelve Data rate-throttles; Oanda has no client throttle)
|
||||
→ engine.TechnicalEngine.analyze() per TF: 9 scored indicators + ATR context → regime → composite score
|
||||
→ signal_aggregator.analyze_pair() live entry point (weekend/day filters, fetch, staleness)
|
||||
└─ analyze_from_dataframes() replayable core: MTF agreement → news → bonuses/penalties →
|
||||
@@ -53,11 +54,11 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz
|
||||
|
||||
- **`providers/`** — one class per provider subclassing `BaseDataProvider` (`providers/base.py`), which owns everything the pipeline depends on: TTL+boundary cache, forming-candle drop, freshness check, validate/clean, paginated `fetch_range` (feeds the backtester from any provider), optional tick streaming. Subclasses implement only `_fetch_raw` + symbol/interval maps. Registry + `make_data_provider(config, name=None)` in `providers/__init__.py`; config.json `data_providers.active` selects the live candle feed and the optional `data_providers.streaming` scalar selects a different provider for the tick stream. A test enforces that subclasses don't override shared base methods (`supports_streaming`/`iter_ticks` are the sanctioned streaming override points). Tiingo streams mid-price ticks (`wss://api.tiingo.com/fx`) — do not re-add spread display from its free FX feed (two-tier indicative quotes, removed 2026-07-23); the no-spread rule extends to the WS stream.
|
||||
- **`engine.py`** — `TechnicalEngine`: 9 scored indicators (ema_trend, rsi, macd, bollinger, adx, roc, keltner, divergence, candlestick) + ATR as context. Regime classification (trending_up/down, `trending` = mixed/direction-UNKNOWN, transitioning, ranging) drives per-indicator weight multipliers and regime-aware RSI reinterpretation. Compute order matters (ATR first; MACD always computed even when not a pillar — divergence needs the histogram). Private series ride on the result as `_underscore` attrs; scorers/formatters skip anything `_`-prefixed or weight-0.
|
||||
- **`signal_aggregator.py`** — MTF agreement, diminishing-decay penalty system, exhaustion hard block, quality grading, then an ordered chain of post-grade gates (session, XAU SELL quality/vol/hour, regime blocks, M5 noise gates), then three setup generators: XAU ATR (A+/A), generic swing-based, plus the M5 pullback (`XAU_PB`) and chart-pattern (`XAU_CP`, `chart_patterns.py`) entry paths. `TradeSetup.strategy` selects trailing params; trail params are stamped on the setup from the *producing* config so per-TF overrides reach formatters.
|
||||
- **`signal_aggregator.py`** — MTF agreement, diminishing-decay penalty system, exhaustion hard block, quality grading, then an ordered chain of post-grade gates (session, XAU SELL quality/vol/hour, regime blocks, M5 noise gates), then three setup generators: XAU ATR (A+/A), generic swing-based, plus the M5 pullback (`XAU_PB`) and chart-pattern (`XAU_CP`, `chart_patterns.py`) entry paths. The PB/CP generators run AFTER the gate chain and force quality to "B"; the CP path deliberately accepts grade-F composites and re-checks only session/day/big-candle (via warning prefixes) plus its own SELL/vol/RSI gates — it does NOT re-check exhaustion or the M5 min-score/MTF gates. **This is validated behavior, not a bug** (2026-08-04 audit: 226/586 trades in the shipped M5 12mo book are XAU_CP, +38.9R, 104 of them exhaustion-flagged and net-positive) — changing it is a rule-1 revalidation event. `TradeSetup.strategy` selects trailing params; trail params are stamped on the setup from the *producing* config so per-TF overrides reach formatters.
|
||||
- **`pair_signals/`** — per-pair config factories. **Always use the factory (`produce_signal` / `make_config`), never bare `PlatformConfig()`** — the factory applies the config.json overlay first, then the validated per-TF deltas (M5 hour-9 block, M15 trail arm 0.70, M15 SELL-hour block, etc.; evidence in the docstring). A test enforces factory ≡ inline-config equivalence. Confirmation map: M5→[M15,H1], M15→[M5,H1], H1→[M15,H4], H4→[H1,D1].
|
||||
- **`news_engine.py`** — Forex Factory calendar (MyFXBook RSS fallback) + RSS keyword sentiment with gold-inverse keywords. Class-level caches (fresh engine per scan). Optional DeepSeek enrichment via `llm_client.py` (requests-only, no SDK): batched headline sentiment overriding keyword scores, gold-direction on release surprises, and a market-summary blurb (`SignalReport.news_summary`) — **display/warning-only, keyword fallback on any failure, inert without a key**; prompts are static module constants (DeepSeek prefix-cache pricing) and requests send `thinking: disabled` (V4-Flash reasoning otherwise blows the timeout). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`) — the LLM does not change that.
|
||||
- **`telegram_bot.py`** — owner-locked bot + subscriber channel; candle-boundary auto-scan loops (M5/M15/H1/H4, boundary+10 s), alert quality threshold + fingerprint dedup (committed only after a successful send), 📰 LLM market-context line on alerts when available, WebSocket level-breach monitor fed by the streaming provider (`data_providers.streaming`, falling back to the active provider; auto-disabled when the resolved provider can't stream or has no API key — Oanda does not stream, Tiingo does). Rotating log: `logs/telegram.log`, 3-h rotation × 240 backups.
|
||||
- **`oanda_daemon.py` / `oanda_executor.py`** — practice-account auto-execution. Book-parity rules (see daemon docstring): no broker-side TP (trailing model — TP1/TP2 advisory), 5-bar per-TF cooldown, trail ratchets at bar close exactly once per bar. Accepted divergence: one position total.
|
||||
- **`oanda_daemon.py` / `oanda_executor.py`** — practice-account auto-execution. Book-parity rules (see daemon docstring): no broker-side TP (trailing model — TP1/TP2 advisory), 5-bar per-TF cooldown (wall-clock approximation, so it expires over weekends where the backtester would still count bars), trail ratchets at bar close once per *observed* bar (a missed/failed cycle skips those bars entirely — the intra-gap high/low never enters the peak). Accepted divergence: one position total; orders send `positionFill=OPEN_ONLY` so a state/broker divergence cancels the order instead of silently netting. State/broker re-sync runs at boot and after any suspicious fill (`_reconcile_positions`).
|
||||
- **`dashboard_publisher.py` / `static/dashboard.html`** — web dashboard as a **static publisher** (live LAMP server): standalone daemon scans at candle closes via `produce_signal` and atomically writes `snapshot.json`/`news.json`/`dashboard.html` to `$DASHBOARD_OUT_DIR` (Apache serves them; page is public by owner decision). **Hard independence rule: never imports, and is never imported by, `telegram_bot`/`oanda_daemon`** — confirmation map and boundary math are deliberate local copies, so a web-side failure can only mean a stale page. Due-TF classification uses the RAW wake time (flooring wake−buffer skips every H1/H4 scan — 2026-08-02 bug). News panel reads `NewsEngine.get_fundamental_context()` plus the additive `get_week_events()` accessor (display-only; no signal path calls it); as a separate process it has its own news caches and DeepSeek daily cap. Deployment checklist: `deploy/README-dashboard.md`.
|
||||
- **`backtest/gpu_backtest.py`** — bar-by-bar replay through the real `analyze_from_dataframes` (news off). Realism model (do not regress): next-bar-open entries re-anchored to the raw price path, round-trip `--spread` charged per trade, trade resolution on RAW bars (cleaning applies only to signal-side data), same-bar SL+TP tie → SL first, gap fills at bar open, trail ratchet at bar close only. Indicators are precomputed once per TF (CuPy GPU or NumPy fallback — check the `Backend:` log line); replay fans out over worker processes and the 5-bar cooldown is applied in an ordered pass afterwards, so parallel and sequential books are identical. Parquet cache in `backtest/cache/` keyed by provider/symbol/TF/window. `backtest/*.csv` are gitignored pinned trade books from past campaigns — used for manual before/after comparison. `backtest/instructions_for_backtest.txt` lists what every backtest report must record.
|
||||
|
||||
@@ -73,6 +74,13 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz
|
||||
8. Don't add config fields nothing reads; don't disable indicators in production paths (`disabled_indicators` is for experiments only).
|
||||
9. EUR/USD was removed 2026-07-16 after a no-edge verdict (cost-to-volatility; last commit with EUR support: `7d3fd95`). Do not re-add without new data. BTC/ETH work lives on unmerged branches only.
|
||||
|
||||
## 2026-08-04 QA audit — owner decisions ledger
|
||||
|
||||
- **Weekend calendar: FIXED (owner-approved 2026-08-04).** All three `_is_forex_weekend` copies are NY-anchored (Fri 17:00 → Sun 18:00 ET; during US DST that is Fri 21:00 UTC — measured on 5yr of Oanda bars). Kills the weekly false-"STALE DATA/feed lagging" hours. Book impact, fully accounted: M15 byte-identical; M5 12mo dropped exactly ONE trade — a phantom whose signal fired on the final summer-Friday bar (close 21:00 UTC) and "filled" at Sunday's reopen 49 h later (−1.08R, live-impossible). M5 12mo reference at HEAD: 555 trades, +98.64R PF≈1.50 @ $0.65 (was 556/+97.56/1.48). A Module 21 test sweeps all three copies hourly across both DST transitions.
|
||||
- **Spread in exit-trigger detection (rule 6): owner DECLINED 2026-08-04.** Backtest stop triggers stay on raw mid High/Low; the measured optimism (~4–6% of total R: −3.55R @ $0.30, −4.24R @ $0.65 on the M15 12mo book) is accepted and books stay comparable. Re-open only as a conscious re-pin campaign.
|
||||
- **`FB2_EMA_LO`: HARDENED 2026-08-04 (delegated).** Research override for the EMA200-proximity band, default = tuned 0.5; an unparseable value now degrades to the default with a warning instead of crashing every scan. Documented here — it is a knob that exists.
|
||||
- **H4 seeding divergence: parked.** Live H4 fetches only 380 bars vs the backtest's unbounded history — ~2% of sampled H4 bars flip an indicator signal (composite Δ up to 12.3). M5/M15/H1 clean at their real fetch depths. Fold into any future H4 campaign; H4 remains "unproven for auto-execution".
|
||||
|
||||
## Current state notes (2026-07-30)
|
||||
|
||||
- Oanda v20 is the live feed (`data_providers.active: "oanda"`), but the pinned backtest books are still Twelve-Data-pinned and the backtest cost assumption is $0.30 while measured real spreads are ~$0.65–0.72 — re-pin books on Oanda bars before treating feed-switch comparisons as valid (see Oanda/TD divergence study in memory).
|
||||
|
||||
@@ -1,85 +0,0 @@
|
||||
# Full Migration to Oanda v20 as Market-Data Source — Roadmap
|
||||
|
||||
**Status: partially executed (updated 2026-08-03).** Phases 1–2 shipped 2026-07-28 via the `providers/` package refactor and the live feed switched to Oanda (`data_providers.active: "oanda"`). Phases 3–6 remain unbuilt: the level-breach stream monitor is simply disabled for Oanda, and the pinned backtest books are still Twelve-Data-pinned at the $0.30 cost assumption. Owner decision 2026-07-24 stands: Twelve Data is a **permanent fallback** (no removal phase).
|
||||
|
||||
## Context
|
||||
|
||||
The daemon already executes trades on an Oanda practice account, and since 2026-07-28 the signal feed (Telegram bot, CLI, backtester) also runs on Oanda. Oanda's API is free with no metering (120 req/s vs TD's 55 credits/min), serves real bid/ask spreads, and matches the actual trading calendar. The 2026-07-24 divergence study (`backtest/research/oanda_td_divergence.py`) proved the switch is **not plumbing-only**: steady-state bars differ by median $0.69 (11% of M15 ATR), bar direction agrees only 89.9%, TD prints ~4,000 phantom Saturday bars/yr + 21:00 UTC rollover bars that contaminate Monday indicator windows, and Oanda's real spread (~$0.72 liquid / $0.84 rollover) is ~2.4× the $0.30 the validated books assume. So the migration = plumbing (Phases 1–3, inert) + shadow evidence (4) + book re-validation with owner gates (5) + reversible cutover (6). CLAUDE.md rules 1 (no tuned-value change without re-validation) and 4 (live == backtest parity) govern throughout.
|
||||
|
||||
**Master sequencing rule:** Phases 1–3 land inert behind `data_providers.active == "twelve_data"`. Nothing changes live behavior until the Phase 6 config flip, which stays reversible forever (TD is a permanent fallback).
|
||||
|
||||
## Phase 1 — Provider plumbing (inert)
|
||||
|
||||
Files: `oanda_provider.py`, `data_provider.py`, `config.py`, `config.json.example`.
|
||||
|
||||
1. **Constructor self-resolution**: `OandaDataProvider.__init__(config, api_token=None, base_url=None)` — defaults resolve via existing `app_config.oanda_api_token()` / `oanda_base_url()`; explicit args keep `oanda_daemon.py:180` working. Fail lazily at first fetch (mirror ForexDataProvider's `_td is None` pattern) so credential-less construction doesn't break the hermetic test header.
|
||||
2. **Registry**: add `"oanda": OandaDataProvider` to `PROVIDER_REGISTRY` (`data_provider.py:392`); import at bottom of `data_provider.py` (one-direction edge, no cycle — `oanda_provider` must never import from `data_provider`). Add `PROVIDER_NAME` class attr to both providers. Cutover config needs an `"oanda": {}` block under `data_providers` (creds stay in the existing top-level `oanda` section / env).
|
||||
3. **Dedupe shared constants**: move `CANDLE_SECONDS` / `BARS_PER_DAY` to `config.py`, alias as class attrs in both providers — all existing call sites (`signal_aggregator.py:18,217,640,942`, `telegram_bot.py:1062`, `gpu_backtest.py:975,1706`, tests) keep working unchanged. Also move `_is_forex_weekend` to `config.py` eventually (currently duplicated + sync-tested).
|
||||
4. **`_fetch_window(symbol, tf, start, end)`** on both providers — the backtester's common interface. Contract: raw closed bars labeled in [start, end], oldest-first, capitalized OHLCV, UTC index; empty DataFrame (not exception) for empty windows; caller keeps windows ≤ ~4500 bars. TD: thin wrapper over existing `_fetch_twelve_data(start_date=, end_date=)` (keep its exception behavior byte-identical). Oanda: new `from`/`to` RFC3339 request (`price=MBA, dailyAlignment=0, alignmentTimezone=Etc/UTC`); refactor `_fetch_oanda_candles`'s parsing body (complete-filter, mid extraction, spread cache) into a shared `_parse_candles()`. The from+count quirk (Oanda's `count` includes the boundary bar; see `backtest/research/fetch_oanda_m15.py`) does not apply to from/to windows. Add a no-op-by-construction `_drop_forming_candle` to OandaDataProvider (`gpu_backtest.py:1025` calls it unconditionally). No throttle needed for Oanda.
|
||||
5. **Tests (extend Module 15)**: registry selection; constructor self-resolution + explicit-args-win; `_fetch_window` both providers (mock `_session.get` / `_fetch_twelve_data`); CANDLE_SECONDS/BARS_PER_DAY cross-provider sync; provider-contract hasattr test covering everything callers reach for (`fetch_ohlcv, get_pair_info, clear_cache, _is_data_fresh, _has_latest_closed_bar, _validate_and_clean, _drop_forming_candle, _fetch_window, CANDLE_SECONDS, BARS_PER_DAY, _cache, _cache_timestamps, PROVIDER_NAME`); port Module 11's two cache/boundary tests (`test_all.py:2278,2313`) to Oanda variants patching `_fetch_oanda_candles`. Tests that swap `set_app_config` must restore `_TEST_APP` in `finally`.
|
||||
|
||||
## Phase 2 — Backtester provider-agnosticism
|
||||
|
||||
File: `backtest/gpu_backtest.py` (`fetch_history`, lines 943–1036).
|
||||
|
||||
1. Line 984: `provider._fetch_twelve_data(...)` → `provider._fetch_window(...)`. Keep the backward-stepping chunk loop, `empty_streak` bail, cursor stepping byte-identical (rule 6 — this loop produced every pinned cache).
|
||||
2. Lines 975/1706: `ForexDataProvider.BARS_PER_DAY` → `provider.BARS_PER_DAY`.
|
||||
3. **Provider-namespaced cache keys** in `_cache_path` (line 938): TD keeps legacy names (pinned parquets stay loadable, needed for A/B), Oanda gets `oanda_` prefix (`oanda_XAU_USD_15min_....parquet`) — no collisions. Comment the asymmetry.
|
||||
4. `clean=True/False` raw-path semantics untouched.
|
||||
5. **Verification before merge**: one M15 12-month TD replay pre/post commit — trade books must be identical. **Alignment audit**: fetch one month of Oanda H4 + D1 via `_fetch_window`, assert label sets match TD's cached labels (H4 at 00/04/08 UTC, D1 at 00:00 UTC) — a mismatch would silently shift every MTF confirmation.
|
||||
6. Tests: `fetch_history` dispatch with a stub provider (concat/dedupe/trim, clean=False skips cleaning); `_cache_path` namespacing regression.
|
||||
|
||||
## Phase 3 — Telegram bot: TD WebSocket → Oanda pricing stream
|
||||
|
||||
File: `telegram_bot.py`; new resolver in `app_config.py`.
|
||||
|
||||
1. New `oanda_stream_url()` resolver: env `OANDA_STREAM_URL` → config.json `oanda.stream_url` (add to `_KNOWN_OANDA`) → derived `oanda_base_url().replace("api-", "stream-", 1)`.
|
||||
2. Pure parser `_parse_stream_line(line) -> Optional[(pair_symbol, mid_price)]`: type==PRICE → mid from closeoutBid/Ask (fallback bids[0]/asks[0]); HEARTBEAT/garbage → None. `_STREAM_TO_PAIR` from `OandaDataProvider.OANDA_INSTRUMENT` (replaces TD-display-keyed `_WS_TO_PAIR`).
|
||||
3. Reader thread (no new deps): `requests.get(.../v3/accounts/{id}/pricing/stream?instruments=..., stream=True, timeout=(10,30))`, iterate `iter_lines(chunk_size=1)` — **chunk_size=1 is load-bearing** (urllib3 buffering vs 5s server heartbeats); push lines to asyncio queue via `loop.call_soon_threadsafe`; interrupt a blocked read with `resp.close()` from the async side.
|
||||
4. Async monitor `_oanda_stream_monitor`: dedicated thread (NOT the 4-worker scan executor), `await asyncio.wait_for(queue.get(), timeout=15)` as staleness watchdog (3× server heartbeat) → reconnect with existing exponential-backoff pattern. PRICE lines feed the identical existing body: `_check_level_breach` → `_handle_breach` → M15 rescan, 5-min cooldown. No client heartbeat.
|
||||
5. **Provider-gated selection** in `_start_scheduler` (line 889): `active_provider == "oanda"` → stream monitor, else existing `_ws_monitor`. This is the rollback hinge and stays permanently (TD fallback). `_stop_scheduler` must stop the reader thread. Generalize `/status` labels (`_status_text` iterates `SCAN_PAIRS`, not `_WS_TO_PAIR`).
|
||||
6. Tests: `_parse_stream_line` variants (PRICE/closeout, PRICE/arrays, HEARTBEAT, malformed, unknown instrument); `_STREAM_TO_PAIR` round-trip; monitor selection under both `active` values; watchdog-reconnect (structure one connection lifecycle as a testable inner coroutine).
|
||||
|
||||
## Phase 4 — Shadow / parity evidence (3–4 weeks, no hot-path code)
|
||||
|
||||
The daemon already runs the full signal stack on Oanda bars. Formalize evidence before any cutover:
|
||||
|
||||
1. **Code parity (exact)**: replay the daemon's shadow window through the backtester on Oanda data (Phase 2 enables this); diff against daemon-logged decisions per (tf, bar_close_ts) — signal, quality, direction, entry/SL/TP. Acceptance: 100% modulo documented daemon divergences (single-position rule, quality gate). Any mismatch is a bug, not feed noise.
|
||||
2. **Feed divergence (measured, not zero-gated)**: `backtest/research/shadow_compare.py` parses `logs/telegram.log` + daemon log into per-scan rows: signal agreement rate, direction flips, grade drift, clustering by hour/weekday. Expect high-single-digit % divergence concentrated near grade thresholds and Monday sessions (TD Saturday/rollover bars). Deliverable: written note attributing every divergence class.
|
||||
3. **Operational telemetry**: boundary grace-retry frequency (is Oanda's just-closed bar present at boundary+10s?), `get_spread()` time series (live input to the Phase 5 cost decision), stream disconnect counts, Monday-open indicator snapshots TD-vs-Oanda.
|
||||
4. **Owner note**: at cutover the data feed host family should match the execution host (practice vs live pricing can micro-differ).
|
||||
|
||||
## Phase 5 — Re-validation campaign (rule 1, owner gates)
|
||||
|
||||
Fetch Oanda history (M5/M15/H1/H4/D1, 2020-07 → present; Oanda XAU reaches ~2005). Run matrix per primary TF (M5/M15/H1, current factory configs): 12-mo train/val halves, 5-yr, OOS 2024-06→2025-03 — each at `--spread 0.30` (pure feed-effect A/B vs pinned TD books), `0.70` (honest Oanda cost), `0.85` (rollover stress). Caution: Jul–Nov 2025 is provider-sensitive (weekly offsets −$6..+$18.5) — lean on 5-yr + OOS for decisions. Acceptance unchanged: positive both halves, PF ≥ 1.3, trailing exits, next-open fills, OOS positive.
|
||||
|
||||
**Owner decision gates (never silent):**
|
||||
1. **Canonical cost** — (a) `round_trip_cost=0.70` everywhere (moves the live TP1 spread gate $9→$21 — a strategy re-tune); (b) `default_spread=0.70` for backtest PnL but retune `min_tp_cost_multiple` 30→~13 to preserve the validated $9 gate; (c) keep $0.30 with documented optimism (only if $0.70 books pass anyway). Full acceptance matrix at chosen values.
|
||||
2. **Degradation handling** per failing TF: re-run tuning grids on Oanda data / drop TF from `scanning.timeframes` / keep Oanda execution-only (status quo).
|
||||
3. **Re-pin**: passing books become the new pinned CSVs + factory-docstring evidence; CLAUDE.md tuned-value annotations update in the cutover commit, not before.
|
||||
|
||||
Files (per decisions): `pair_signals/xauusd.py`, `config.py` (`PAIR_INFO.default_spread`, `SignalConfig.round_trip_cost`/`min_tp_cost_multiple`), `test_all.py` tuned-constant assertions, CLAUDE.md.
|
||||
|
||||
## Phase 6 — Cutover (reversible forever; no TD removal)
|
||||
|
||||
1. config.json: `data_providers.active: "oanda"` + `"oanda": {}` block; restart bot. Everything flips through two seams: `make_data_provider` (bot shared dp, CLI, backtester) and the Phase 3 monitor selector. Any Phase 5 tuned-value changes land in this commit with their validation records.
|
||||
2. **Bake ≥1 week** (full weekly cycle incl. Monday open): grace-retry rate, stream stability, alert plausibility vs daemon, `/cache` freshness.
|
||||
3. **Rollback = flip `active` back** — TD code, key, caches, and WS monitor all stay maintained indefinitely (owner decision). Dual-provider posture: cross-provider sync tests (Phase 1) keep both honest; TD subscription/key must stay valid to remain a real fallback.
|
||||
4. Docs: CLAUDE.md (pipeline header, data_provider section, env-var table + `OANDA_STREAM_URL`, backtester cache note, comparability note: "pre-Oanda books are TD-feed books"), `docs/` guides if they name the data source.
|
||||
|
||||
## Key risks
|
||||
|
||||
1. **Cost-gate coupling**: naively setting `round_trip_cost=0.70` silently moves the live $9 TP1 gate to $21 — owner gate 5.1 exists precisely for this.
|
||||
2. **Cutover before re-pinning** — feed divergence is material (11% of ATR, 10% direction flips); sequencing rule prevents it.
|
||||
3. **Hermetic test header** — `_TEST_APP` pins twelve_data; provider-swapping tests must restore it or later modules fail mysteriously.
|
||||
4. **Stream liveness triad** — `chunk_size=1`, `resp.close()` interruption, 15s watchdog.
|
||||
5. **H4/D1 alignment** — audit before trusting any Oanda backtest.
|
||||
6. **Monday/rollover divergence class** — attribute in Phase 4 evidence; do not "fix".
|
||||
|
||||
## Verification (per phase)
|
||||
|
||||
- Phases 1–3: `python test_all.py` green (235 + new); live behavior provably unchanged (TD active).
|
||||
- Phase 2: pre/post TD book identity diff; Oanda H4/D1 alignment audit.
|
||||
- Phase 3: soak a second bot instance with `active=oanda` scratch config; observe stream reconnects in log.
|
||||
- Phase 5: full acceptance matrix, owner sign-offs recorded in factory docstrings.
|
||||
- Phase 6: 1-week bake checklist above; rollback drill (flip back, confirm TD scan cycle) once during bake.
|
||||
+106
-20
@@ -40,10 +40,13 @@ from dataclasses import dataclass, field
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
_NY_TZ = ZoneInfo("America/New_York")
|
||||
|
||||
# Ensure project root on sys.path so this script can be invoked from anywhere.
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
if str(PROJECT_ROOT) not in sys.path:
|
||||
@@ -402,6 +405,11 @@ class GPUTechnicalEngine(TechnicalEngine):
|
||||
# cache key = timeframe.value (e.g. "15min", "1h", "4h")
|
||||
self.caches = caches
|
||||
self._current_tf: Optional[str] = None
|
||||
# Cache-miss fallbacks recompute with the live engine's pandas
|
||||
# math (different EWM seeding than the precomputed cache). That
|
||||
# is a parity hazard, so it must never happen silently.
|
||||
self.fallback_count = 0
|
||||
self._fallback_warned = False
|
||||
|
||||
def analyze(self, df: pd.DataFrame, symbol: str, timeframe: str) -> TechnicalAnalysis:
|
||||
# Latch the timeframe so per-indicator overrides know which cache to use.
|
||||
@@ -419,9 +427,23 @@ class GPUTechnicalEngine(TechnicalEngine):
|
||||
return None, -1
|
||||
cache = self.caches.get(self._current_tf)
|
||||
if cache is None:
|
||||
self._note_fallback(f"no cache for TF {self._current_tf!r}")
|
||||
return None, -1
|
||||
pos = cache.index_pos.get(df.index[-1])
|
||||
return (cache, pos) if pos is not None else (None, -1)
|
||||
if pos is None:
|
||||
self._note_fallback(
|
||||
f"{self._current_tf}: slice end {df.index[-1]} not in cache")
|
||||
return None, -1
|
||||
return cache, pos
|
||||
|
||||
def _note_fallback(self, why: str) -> None:
|
||||
self.fallback_count += 1
|
||||
if not self._fallback_warned:
|
||||
self._fallback_warned = True
|
||||
logger.warning(
|
||||
"GPU cache miss → falling back to live-engine recompute "
|
||||
"(%s). Further misses counted silently; check "
|
||||
"fallback_count at end of run.", why)
|
||||
|
||||
# ── EMA overrides ──────────────────────────────────────────────
|
||||
|
||||
@@ -826,7 +848,12 @@ class GPUTechnicalEngine(TechnicalEngine):
|
||||
c = self.config
|
||||
n = c.roc_period
|
||||
|
||||
if pos < n + 1:
|
||||
# Mirror the live guard (engine._compute_roc: len < n + 5 → skip,
|
||||
# i.e. emit only from pos >= n + 4). Unreachable in practice —
|
||||
# analyze() floors at 30 bars and n+4 < 30 for shipped periods —
|
||||
# but keep the engines textually aligned so a future period/floor
|
||||
# change can't open a silent divergence window.
|
||||
if pos < n + 4:
|
||||
return
|
||||
|
||||
roc_val = float(cache.roc[pos])
|
||||
@@ -876,6 +903,11 @@ class GPUTechnicalEngine(TechnicalEngine):
|
||||
return super()._compute_keltner(df, result)
|
||||
|
||||
c = self.config
|
||||
# Mirror the live guard (engine._compute_keltner: len < max(kc_period,
|
||||
# kc_atr_period) + 5 → skip). Unreachable below the 30-bar analyze()
|
||||
# floor with shipped periods; kept aligned for the same reason as ROC.
|
||||
if pos < max(c.kc_period, c.kc_atr_period) + 4:
|
||||
return
|
||||
latest = float(df["Close"].iloc[-1])
|
||||
kc_upper_val = float(cache.kc_upper[pos])
|
||||
kc_lower_val = float(cache.kc_lower[pos])
|
||||
@@ -979,6 +1011,16 @@ def fetch_history(
|
||||
path = _cache_path(symbol, timeframe, start, end, provider.name)
|
||||
if use_cache and path.exists():
|
||||
logger.info(f"Loading cached {symbol} {timeframe.value}: {path.name}")
|
||||
# Files written before the raw-cache change (2026-07-08, commit
|
||||
# 6a96ed6) contain CLEANED bars; loading one with clean=False
|
||||
# would silently miss the spike bars the raw path exists to model.
|
||||
# The key format predates versioning, so detect by mtime.
|
||||
_RAW_CACHE_EPOCH = 1783468800.0 # 2026-07-08 00:00 UTC
|
||||
if path.stat().st_mtime < _RAW_CACHE_EPOCH:
|
||||
logger.error(
|
||||
f"{path.name} predates the raw-cache format (2026-07-08) "
|
||||
f"and likely contains CLEANED bars — delete it and refetch "
|
||||
f"before trusting trade resolution on this window")
|
||||
df = pd.read_parquet(path)
|
||||
if df.index.tz is None:
|
||||
df.index = df.index.tz_localize("UTC")
|
||||
@@ -1130,6 +1172,15 @@ def _simulate_trailing(
|
||||
reason = "TIMEOUT"
|
||||
for j in range(n):
|
||||
if np.isnan(h[j]):
|
||||
# Data ends here: close at the LAST VALID bar, not at
|
||||
# c[n-1] (which may itself be NaN — a NaN exit price used
|
||||
# to poison r_multiple and every downstream metric).
|
||||
if j == 0:
|
||||
exit_idx = 0
|
||||
reason = "NO_DATA"
|
||||
else:
|
||||
exit_idx = j - 1
|
||||
exit_price = float(c[exit_idx])
|
||||
break
|
||||
hit = (l[j] <= cur_sl) if buy else (h[j] >= cur_sl)
|
||||
if hit:
|
||||
@@ -1151,8 +1202,15 @@ def _simulate_trailing(
|
||||
else peak + dist * risk)
|
||||
cur_sl = max(cur_sl, lvl) if buy else min(cur_sl, lvl)
|
||||
|
||||
if reason == "NO_DATA":
|
||||
t.exit_reason = reason
|
||||
continue
|
||||
if exit_price is None:
|
||||
exit_price = float(c[exit_idx])
|
||||
if not np.isfinite(exit_price):
|
||||
# Last-resort guard: never emit a NaN exit/R into the book.
|
||||
t.exit_reason = "NO_DATA"
|
||||
continue
|
||||
t.exit_reason = reason
|
||||
t.bars_held = exit_idx + 1
|
||||
t.exit_price = float(exit_price)
|
||||
@@ -1325,6 +1383,12 @@ def simulate_trades(
|
||||
last_idx = int(n_valid) - 1
|
||||
t.exit_price = float(closes_arr[i, last_idx])
|
||||
t.exit_time = exit_times[i][last_idx]
|
||||
if not np.isfinite(t.exit_price):
|
||||
# NaN close on the final window bar (corrupt raw data)
|
||||
# must not become a NaN R.
|
||||
t.exit_reason = "NO_DATA"
|
||||
t.exit_price = None
|
||||
continue
|
||||
|
||||
# Realized R from actual fill prices, net of round-trip spread.
|
||||
risk = abs(t.entry_price - t.sl)
|
||||
@@ -1389,18 +1453,22 @@ class BacktestConfig:
|
||||
def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]:
|
||||
"""Apply XAU weekend + allowed-weekday filter matching live `analyze_pair`.
|
||||
|
||||
Forex market closed: Friday 22:00 UTC -> Sunday 22:00 UTC (Sydney open).
|
||||
Must match `_is_forex_weekend` in signal_aggregator.py exactly.
|
||||
Market closed Friday 17:00 -> Sunday 18:00 NEW YORK time (= Fri
|
||||
21:00 UTC in US summer, 22:00 in winter — matches measured Oanda
|
||||
feed behavior; 2026-08-04 audit I43). Must match `_is_forex_weekend`
|
||||
in signal_aggregator.py exactly (rule 4; Module 21 agreement test).
|
||||
Note the weekday gates below stay on UTC weekday — they mirror the
|
||||
live gates, which evaluate UTC weekdays.
|
||||
"""
|
||||
xau = cfg.xau_strategy
|
||||
if xau.weekend_filter:
|
||||
wd = ts.weekday()
|
||||
hour = ts.hour
|
||||
if wd == 4 and hour >= 22:
|
||||
ny = ts.astimezone(_NY_TZ)
|
||||
wd = ny.weekday()
|
||||
if wd == 4 and ny.hour >= 17:
|
||||
return "WEEKEND_FILTER"
|
||||
if wd == 5:
|
||||
return "WEEKEND_FILTER"
|
||||
if wd == 6 and hour < 22:
|
||||
if wd == 6 and ny.hour < 18:
|
||||
return "WEEKEND_FILTER"
|
||||
if xau.allowed_weekdays and ts.weekday() not in xau.allowed_weekdays:
|
||||
return "DAY_FILTER"
|
||||
@@ -1617,6 +1685,24 @@ def _reanchor_entries_to_raw(
|
||||
)
|
||||
|
||||
|
||||
def _apply_cooldown(
|
||||
candidates: List[Tuple[int, Trade]], cooldown_bars: int
|
||||
) -> List[Tuple[int, Trade]]:
|
||||
"""Ordered cooldown pass over (bar_pos, Trade) candidates: drop any
|
||||
candidate within `cooldown_bars` bars of the previously ACCEPTED
|
||||
entry — exactly what a sequential replay loop would have skipped.
|
||||
Candidates must already be sorted by position (stable merge from the
|
||||
worker spans)."""
|
||||
accepted: List[Tuple[int, Trade]] = []
|
||||
last_entry_pos = -10_000
|
||||
for pos, trade in candidates:
|
||||
if pos - last_entry_pos < cooldown_bars:
|
||||
continue
|
||||
accepted.append((pos, trade))
|
||||
last_entry_pos = pos
|
||||
return accepted
|
||||
|
||||
|
||||
def run_backtest(
|
||||
bt_cfg: BacktestConfig,
|
||||
platform_cfg: Optional[PlatformConfig] = None,
|
||||
@@ -1744,18 +1830,10 @@ def run_backtest(
|
||||
candidates.extend(cands)
|
||||
candidates.sort(key=lambda pc: pc[0])
|
||||
|
||||
# ── Phase 2: entry cooldown — the only sequential rule. A candidate
|
||||
# inside the cooldown window of the previously ACCEPTED entry is
|
||||
# dropped, exactly as the sequential loop skipped those bars.
|
||||
trades: List[Trade] = []
|
||||
signal_ts_list: List[pd.Timestamp] = []
|
||||
last_entry_pos = -10_000
|
||||
for pos, trade in candidates:
|
||||
if pos - last_entry_pos < bt_cfg.cooldown_bars:
|
||||
continue
|
||||
trades.append(trade)
|
||||
signal_ts_list.append(primary_df.index[pos])
|
||||
last_entry_pos = pos
|
||||
# ── Phase 2: entry cooldown — the only sequential rule.
|
||||
accepted = _apply_cooldown(candidates, bt_cfg.cooldown_bars)
|
||||
trades: List[Trade] = [t for _, t in accepted]
|
||||
signal_ts_list: List[pd.Timestamp] = [primary_df.index[p] for p, _ in accepted]
|
||||
|
||||
logger.info(
|
||||
f"Replay complete: {len(trades)} entries in "
|
||||
@@ -1828,6 +1906,14 @@ def _summarize(rs: List[float]) -> Dict:
|
||||
|
||||
def compute_metrics(trades: List[Trade]) -> Metrics:
|
||||
resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"]
|
||||
# Last line of defense: one non-finite R (NaN exit on corrupt data)
|
||||
# would silently NaN every aggregate below. Exclude loudly.
|
||||
bad = [t for t in resolved if not np.isfinite(t.r_multiple)]
|
||||
if bad:
|
||||
logger.error(
|
||||
"compute_metrics: excluding %d trade(s) with non-finite R "
|
||||
"(corrupt exit data) — first at %s", len(bad), bad[0].entry_time)
|
||||
resolved = [t for t in resolved if np.isfinite(t.r_multiple)]
|
||||
if not resolved:
|
||||
return Metrics(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, {}, {}, {})
|
||||
|
||||
|
||||
@@ -231,8 +231,11 @@ def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
||||
elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
|
||||
interval_seconds = interval_minutes * 60
|
||||
remaining = interval_seconds - elapsed_seconds + buffer
|
||||
if remaining <= buffer:
|
||||
remaining += interval_seconds
|
||||
# NOTE: remaining > buffer always (elapsed < interval by construction);
|
||||
# a call at the exact boundary instant targets the NEXT boundary —
|
||||
# steady-state loops wake at boundary+buffer, so only a process
|
||||
# started in the boundary microsecond skips one scan (2026-08-04
|
||||
# audit I42; the old `remaining <= buffer` guard was unreachable).
|
||||
return remaining
|
||||
|
||||
|
||||
|
||||
+94
-15
@@ -90,8 +90,17 @@ STATE_FILE = os.path.join(
|
||||
def _oanda_setting_int(env_key: str, json_key: str, default: int) -> int:
|
||||
env = os.environ.get(env_key)
|
||||
if env:
|
||||
try:
|
||||
return int(env)
|
||||
except ValueError:
|
||||
raise SystemExit(
|
||||
f"{env_key}={env!r} is not an integer — refusing to start")
|
||||
try:
|
||||
return int(get_app_config().oanda_settings.get(json_key, default))
|
||||
except (TypeError, ValueError):
|
||||
raise SystemExit(
|
||||
f"config.json oanda.{json_key} is not an integer — "
|
||||
f"refusing to start")
|
||||
|
||||
|
||||
def _oanda_setting_str(env_key: str, json_key: str, default: str) -> str:
|
||||
@@ -130,8 +139,11 @@ def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
||||
elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
|
||||
interval_seconds = interval_minutes * 60
|
||||
remaining = interval_seconds - elapsed_seconds + buffer
|
||||
if remaining <= buffer:
|
||||
remaining += interval_seconds
|
||||
# NOTE: remaining > buffer always (elapsed < interval by construction);
|
||||
# a call at the exact boundary instant targets the NEXT boundary —
|
||||
# steady-state loops wake at boundary+buffer, so only a process
|
||||
# started in the boundary microsecond skips one scan (2026-08-04
|
||||
# audit I42; the old `remaining <= buffer` guard was unreachable).
|
||||
return remaining
|
||||
|
||||
# ── Trade state persistence ───────────────────────────────────────────
|
||||
@@ -165,7 +177,13 @@ class TradingDaemon:
|
||||
base_url = oanda_base_url()
|
||||
lot_size = _oanda_setting_int("OANDA_LOT_SIZE", "lot_size", 1)
|
||||
self._quality_gate = _oanda_setting_str(
|
||||
"OANDA_QUALITY_GATE", "quality_gate", "C")
|
||||
"OANDA_QUALITY_GATE", "quality_gate", "C").strip()
|
||||
if self._quality_gate not in QUALITY_RANK:
|
||||
# Fail CLOSED: an unrecognized gate must never silently fall
|
||||
# back to the most permissive rank (QUALITY_RANK.get default).
|
||||
raise SystemExit(
|
||||
f"OANDA_QUALITY_GATE/quality_gate {self._quality_gate!r} "
|
||||
f"is not one of {sorted(QUALITY_RANK)} — refusing to start")
|
||||
self._max_hold_bars = _oanda_setting_int(
|
||||
"OANDA_MAX_HOLD_BARS", "max_hold_bars", 300)
|
||||
self._price_precision = PAIR_INFO[SYMBOL]["price_precision"]
|
||||
@@ -275,19 +293,30 @@ class TradingDaemon:
|
||||
# ── Position discovery (restart recovery) ──────────────────────
|
||||
|
||||
def _discover_existing_positions(self) -> None:
|
||||
self._reconcile_positions("boot discovery")
|
||||
|
||||
def _reconcile_positions(self, reason: str) -> None:
|
||||
"""Re-sync tracked state with the broker's open trades. Safe to
|
||||
call mid-run (lock-guarded); used at boot and whenever a fill or
|
||||
placement error suggests state and broker may have diverged."""
|
||||
try:
|
||||
open_trades = self._executor.get_open_trades(INSTRUMENT)
|
||||
except OandaAPIError:
|
||||
logger.exception("Failed to query open trades at boot")
|
||||
logger.exception(
|
||||
"Failed to query open trades (%s) — reconcile skipped",
|
||||
reason)
|
||||
return
|
||||
|
||||
saved = self._state.get("trades", {})
|
||||
live_ids = {t["id"] for t in open_trades}
|
||||
|
||||
with self._state_lock:
|
||||
saved = self._state.get("trades", {})
|
||||
|
||||
stale = [tid for tid in saved if tid not in live_ids]
|
||||
for tid in stale:
|
||||
logger.info(
|
||||
"Stale trade %s in state file (closed on Oanda), removing", tid)
|
||||
"Stale trade %s in state file (closed on Oanda), "
|
||||
"removing", tid)
|
||||
del saved[tid]
|
||||
|
||||
for trade in open_trades:
|
||||
@@ -299,11 +328,26 @@ class TradingDaemon:
|
||||
saved[tid].get("entry"), saved[tid].get("current_sl"))
|
||||
continue
|
||||
|
||||
direction = "BUY" if float(trade.get("currentUnits", 0)) > 0 else "SELL"
|
||||
direction = ("BUY" if float(trade.get("currentUnits", 0)) > 0
|
||||
else "SELL")
|
||||
entry = float(trade.get("price", 0))
|
||||
sl_order = trade.get("stopLossOrder", {})
|
||||
sl = float(sl_order.get("price", entry))
|
||||
risk = abs(entry - sl) if sl else 1.0
|
||||
sl_order = trade.get("stopLossOrder") or {}
|
||||
sl_price = sl_order.get("price")
|
||||
if sl_price is None:
|
||||
# No broker-side stop at all: risk is undefined, so the
|
||||
# trailing model cannot manage this trade. Scream — a
|
||||
# naked position needs a human (or a policy) to set a
|
||||
# stop; fabricating one here would invent trading
|
||||
# behavior the books never validated.
|
||||
logger.critical(
|
||||
"Adopted trade %s (%s @ %.2f) has NO STOP LOSS on "
|
||||
"the broker — it will NOT be trailed. Set a stop "
|
||||
"manually or close it.", tid, direction, entry)
|
||||
sl = entry
|
||||
risk = 0.0
|
||||
else:
|
||||
sl = float(sl_price)
|
||||
risk = abs(entry - sl)
|
||||
|
||||
saved[tid] = {
|
||||
"tf": "H1",
|
||||
@@ -328,7 +372,7 @@ class TradingDaemon:
|
||||
self._state["trades"] = saved
|
||||
_save_state(self._state)
|
||||
logger.info(
|
||||
"Position discovery: %d active trade(s)", len(saved))
|
||||
"Position reconcile (%s): %d active trade(s)", reason, len(saved))
|
||||
|
||||
# ── Candle-boundary loop ───────────────────────────────────────
|
||||
|
||||
@@ -521,9 +565,15 @@ class TradingDaemon:
|
||||
stop_loss=ts.stop_loss,
|
||||
)
|
||||
except OandaAPIError:
|
||||
# The order is never retried (a timeout after a server-side
|
||||
# fill would double-up on retry), but the server MAY have
|
||||
# filled before the error reached us — reconcile so a live
|
||||
# position can't stay untracked until the next restart.
|
||||
logger.exception(
|
||||
"Failed to place %s order for %s %s",
|
||||
"Failed to place %s order for %s %s — reconciling "
|
||||
"broker positions",
|
||||
ts.direction, SYMBOL, tf.value)
|
||||
self._reconcile_positions("order placement error")
|
||||
return
|
||||
|
||||
fill_tx = result.get("orderFillTransaction")
|
||||
@@ -535,7 +585,16 @@ class TradingDaemon:
|
||||
trade_opened = fill_tx.get("tradeOpened", {})
|
||||
trade_id = trade_opened.get("tradeID")
|
||||
if not trade_id:
|
||||
logger.warning("No trade ID in fill response")
|
||||
# A fill with no tradeOpened means the order interacted
|
||||
# with an existing broker position (tradesClosed /
|
||||
# tradeReduced on a netting account) — state and broker
|
||||
# have diverged. Surface it and re-sync instead of
|
||||
# silently forgetting a real fill.
|
||||
logger.critical(
|
||||
"Order FILLED but opened no trade (netting close/"
|
||||
"reduce?) — fill: closed=%s reduced=%s. Reconciling.",
|
||||
fill_tx.get("tradesClosed"), fill_tx.get("tradeReduced"))
|
||||
self._reconcile_positions("fill without tradeOpened")
|
||||
return
|
||||
|
||||
fill_price = float(fill_tx.get("price", ts.entry_price))
|
||||
@@ -587,8 +646,12 @@ class TradingDaemon:
|
||||
# ── Trailing stop management ───────────────────────────────────
|
||||
|
||||
def _manage_trailing(self, tf: Timeframe) -> None:
|
||||
# Copy the per-trade dicts too: the loop mutates info (peak,
|
||||
# bars_held, last_bar_ts) and must not touch live state outside
|
||||
# the lock — another thread may be serializing it in _save_state.
|
||||
with self._state_lock:
|
||||
trades = dict(self._state.get("trades", {}))
|
||||
trades = {tid: dict(info)
|
||||
for tid, info in self._state.get("trades", {}).items()}
|
||||
|
||||
for trade_id, info in trades.items():
|
||||
if info.get("tf") != tf.value:
|
||||
@@ -634,11 +697,14 @@ class TradingDaemon:
|
||||
self._executor.close_trade(trade_id)
|
||||
except OandaAPIError:
|
||||
# Keep tracking — the trade is still live on the
|
||||
# broker; we retry the close at the next bar.
|
||||
# broker (incl. a 200 whose close order was cancelled,
|
||||
# e.g. MARKET_HALTED — close_trade raises on no-fill);
|
||||
# we retry the close at the next bar.
|
||||
logger.exception(
|
||||
"Failed to close timed-out trade %s — will retry",
|
||||
trade_id)
|
||||
with self._state_lock:
|
||||
if trade_id in self._state["trades"]:
|
||||
self._state["trades"][trade_id] = info
|
||||
_save_state(self._state)
|
||||
continue
|
||||
@@ -651,6 +717,14 @@ class TradingDaemon:
|
||||
bar_high = float(last_bar["High"])
|
||||
bar_low = float(last_bar["Low"])
|
||||
|
||||
if info.get("risk", 0) <= 0:
|
||||
# Adopted naked position (no broker SL → risk undefined):
|
||||
# never manageable by the trailing model — keep it loud
|
||||
# instead of silently skipping forever.
|
||||
logger.warning(
|
||||
"Trade %s has risk<=0 (no stop loss) — trailing "
|
||||
"skipped; set a stop manually", trade_id)
|
||||
|
||||
new_sl = self._compute_trail(info, bar_high, bar_low)
|
||||
|
||||
if new_sl is not None:
|
||||
@@ -667,6 +741,11 @@ class TradingDaemon:
|
||||
trade_id)
|
||||
|
||||
with self._state_lock:
|
||||
# Write back ONLY if the trade is still tracked — another
|
||||
# loop's _check_closed_trades may have popped it while we
|
||||
# iterated (broker-side close). Re-inserting would create
|
||||
# a ghost that blocks new entries.
|
||||
if trade_id in self._state["trades"]:
|
||||
self._state["trades"][trade_id] = info
|
||||
_save_state(self._state)
|
||||
|
||||
|
||||
+19
-1
@@ -69,6 +69,13 @@ class OandaExecutor:
|
||||
"instrument": instrument,
|
||||
"units": signed_units,
|
||||
"timeInForce": "FOK",
|
||||
# The book model is one position total; on a netting
|
||||
# account an opposite-side order would silently CLOSE an
|
||||
# existing (possibly untracked) position instead of opening
|
||||
# the signal trade. OPEN_ONLY makes that case fail loudly
|
||||
# (order cancelled) so the divergence surfaces instead of
|
||||
# executing an unintended close.
|
||||
"positionFill": "OPEN_ONLY",
|
||||
"stopLossOnFill": {
|
||||
"price": self._fmt_price(stop_loss),
|
||||
},
|
||||
@@ -156,7 +163,18 @@ class OandaExecutor:
|
||||
f"/trades/{trade_id}/close")
|
||||
resp = self._request_with_retry("PUT", url, json={"units": "ALL"})
|
||||
data = self._parse_json(resp)
|
||||
close_tx = data.get("orderFillTransaction", {})
|
||||
close_tx = data.get("orderFillTransaction")
|
||||
if not close_tx:
|
||||
# A 200 whose close order was created but NOT filled (e.g.
|
||||
# cancelled with MARKET_HALTED during the settlement pause)
|
||||
# means the position is STILL OPEN. Raising keeps the caller
|
||||
# tracking it and retrying — silently returning here would
|
||||
# orphan a live position.
|
||||
reason = data.get("orderCancelTransaction", {}).get(
|
||||
"reason", "no orderFillTransaction in response")
|
||||
raise OandaAPIError(
|
||||
f"Close of trade {trade_id} did not fill: {reason}",
|
||||
resp.status_code, str(data)[:300])
|
||||
close_price = float(close_tx.get("price", 0))
|
||||
pl = float(close_tx.get("pl", 0))
|
||||
logger.info(
|
||||
|
||||
@@ -197,6 +197,13 @@ def make_config(
|
||||
config.xau_strategy.blocked_hours_utc = [
|
||||
h for h in config.xau_strategy.blocked_hours_utc if h != 21
|
||||
] + [9]
|
||||
# Deliberately NOT mirrored into config.backtest.blocked_hours_utc:
|
||||
# that copy is the generic non-XAU fallback carrying the JSON base
|
||||
# only. Propagating a gold-validated per-TF delta there would leak
|
||||
# XAU-calibrated hours to a future pair (pair-isolation rule).
|
||||
# Every XAU consumer branches to the xau_strategy copy. Enforced
|
||||
# by the Module 13 test "M5 factory appends hour 9 on top of a
|
||||
# JSON base".
|
||||
# BUY-only chart patterns on M5 — see module docstring.
|
||||
config.xau_strategy.cp_enabled = True
|
||||
config.xau_strategy.cp_patterns = ["inv_hs", "double_bottom", "bull_flag"]
|
||||
|
||||
+138
-29
@@ -24,6 +24,7 @@ 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
|
||||
|
||||
@@ -31,19 +32,36 @@ 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:
|
||||
"""Forex market closed: Friday 22:00 UTC → Sunday 22:00 UTC."""
|
||||
"""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")
|
||||
wd = now.weekday()
|
||||
hour = now.hour
|
||||
if wd == 4 and hour >= 22:
|
||||
return True
|
||||
if wd == 5:
|
||||
return True
|
||||
if wd == 6 and hour < 22:
|
||||
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
|
||||
|
||||
|
||||
@@ -86,9 +104,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,17 +143,28 @@ class BaseDataProvider(ABC):
|
||||
lookback_days = self.config.data_lookback_days
|
||||
|
||||
cache_key = f"{symbol}_{timeframe.value}"
|
||||
now = time.time()
|
||||
outputsize = max(
|
||||
300,
|
||||
min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20,
|
||||
self.MAX_BARS_PER_REQUEST),
|
||||
)
|
||||
|
||||
cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval)
|
||||
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)):
|
||||
@@ -138,21 +173,31 @@ class BaseDataProvider(ABC):
|
||||
with self._cache_lock:
|
||||
self._cache.pop(cache_key, None)
|
||||
self._cache_timestamps.pop(cache_key, None)
|
||||
self._cache_outputsize.pop(cache_key, None)
|
||||
|
||||
outputsize = max(
|
||||
300,
|
||||
min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20,
|
||||
self.MAX_BARS_PER_REQUEST),
|
||||
)
|
||||
# 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.
|
||||
# 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:
|
||||
@@ -180,7 +225,15 @@ class BaseDataProvider(ABC):
|
||||
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 +252,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 +261,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 +289,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 +314,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 +404,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 +464,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 +492,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()
|
||||
|
||||
+18
-4
@@ -149,23 +149,37 @@ class OandaDataProvider(BaseDataProvider):
|
||||
rows = []
|
||||
last_bid_close = None
|
||||
last_ask_close = None
|
||||
skipped_malformed = 0
|
||||
for c in candles:
|
||||
if not c.get("complete", False):
|
||||
continue
|
||||
mid = c.get("mid", {})
|
||||
bid = c.get("bid", {})
|
||||
ask = c.get("ask", {})
|
||||
# A candle missing any OHLC field must be SKIPPED, never
|
||||
# defaulted to 0.0: a zero price flows through cleaning as a
|
||||
# "legit" bar and poisons ATR/Keltner/SL math — and a
|
||||
# mid-frame zero Close makes pct_change() emit inf, which
|
||||
# NaNs the std and disarms the 10σ outlier filter for the
|
||||
# whole frame (2026-08-04 audit, I45).
|
||||
if any(mid.get(k) is None for k in ("o", "h", "l", "c")):
|
||||
skipped_malformed += 1
|
||||
continue
|
||||
rows.append({
|
||||
"Datetime": pd.Timestamp(c["time"]).tz_convert("UTC"),
|
||||
"Open": float(mid.get("o", 0)),
|
||||
"High": float(mid.get("h", 0)),
|
||||
"Low": float(mid.get("l", 0)),
|
||||
"Close": float(mid.get("c", 0)),
|
||||
"Open": float(mid["o"]),
|
||||
"High": float(mid["h"]),
|
||||
"Low": float(mid["l"]),
|
||||
"Close": float(mid["c"]),
|
||||
"Volume": int(c.get("volume", 0)),
|
||||
})
|
||||
if bid and ask:
|
||||
last_bid_close = float(bid.get("c", 0))
|
||||
last_ask_close = float(ask.get("c", 0))
|
||||
if skipped_malformed:
|
||||
logger.warning(
|
||||
"Oanda returned %d malformed candle(s) for %s (missing "
|
||||
"OHLC fields) — skipped", skipped_malformed, instrument)
|
||||
|
||||
if not rows:
|
||||
raise OandaDataError(
|
||||
|
||||
@@ -210,6 +210,12 @@ class TwelveDataProvider(BaseDataProvider):
|
||||
The TD WS speaks display symbols ("XAU/USD"); ticks are mapped back
|
||||
to the canonical pair symbols passed in. The caller owns the
|
||||
reconnect/backoff loop around this generator.
|
||||
|
||||
`on_connect` fires on the FIRST RECEIVED frame, not after the
|
||||
subscribe send (2026-08-04 audit H38b, mirroring the Tiingo
|
||||
provider) — firing pre-receive let a rejected API key reset the
|
||||
caller's reconnect backoff on every attempt, pinning it at the
|
||||
minimum forever.
|
||||
"""
|
||||
import websockets
|
||||
|
||||
@@ -222,11 +228,14 @@ class TwelveDataProvider(BaseDataProvider):
|
||||
}
|
||||
await ws.send(json.dumps(subscribe))
|
||||
logger.info(f"WebSocket connected, subscribed to {len(ws_to_pair)} symbols")
|
||||
if on_connect is not None:
|
||||
on_connect()
|
||||
hb_task = asyncio.create_task(self._ws_heartbeat(ws))
|
||||
connected = False
|
||||
try:
|
||||
async for raw in ws:
|
||||
if not connected:
|
||||
connected = True
|
||||
if on_connect is not None:
|
||||
on_connect()
|
||||
parsed = self._parse_ws_price(raw)
|
||||
if parsed is None:
|
||||
continue
|
||||
|
||||
+25
-8
@@ -10,9 +10,12 @@ import os
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Dict, List, Optional
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import pandas as pd
|
||||
|
||||
_NY_TZ = ZoneInfo("America/New_York")
|
||||
|
||||
from chart_patterns import detect_chart_pattern
|
||||
from config import PlatformConfig, SignalStrength, Timeframe
|
||||
from providers import BaseDataProvider, make_data_provider
|
||||
@@ -23,15 +26,19 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _is_forex_weekend(now_utc: datetime) -> bool:
|
||||
"""Forex market closed: Friday 22:00 UTC → Sunday 22:00 UTC (Sydney open)."""
|
||||
wd = now_utc.weekday()
|
||||
hour = now_utc.hour
|
||||
if wd == 4 and hour >= 22: # Friday after 22:00
|
||||
return True
|
||||
if wd == 5: # All Saturday
|
||||
return True
|
||||
if wd == 6 and hour < 22: # Sunday before 22:00
|
||||
"""Gold/FX market closed for the weekend: Friday 17:00 → Sunday 18:00
|
||||
NEW YORK time (= Fri 21:00 UTC in US summer, 22:00 in winter —
|
||||
matches measured Oanda feed behavior; 2026-08-04 audit I43).
|
||||
Deliberate copy of providers.base._is_forex_weekend — a Module 21
|
||||
test asserts the copies agree across DST transitions."""
|
||||
ny = now_utc.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
|
||||
|
||||
|
||||
@@ -1253,7 +1260,17 @@ class SignalAggregator:
|
||||
ema200 = primary.key_levels.get("ema_trend")
|
||||
if ema200 and ema200 > 0 and report.current_price > 0:
|
||||
ema_dist_pct = abs(report.current_price - ema200) / ema200 * 100.0
|
||||
# FB2_EMA_LO: research override for the EMA200-proximity
|
||||
# penalty band; the tuned default is 0.5. An unparseable
|
||||
# value must degrade to the default, never crash the scan
|
||||
# (2026-08-04 audit; hardened per owner delegation).
|
||||
try:
|
||||
_lo = float(os.environ.get("FB2_EMA_LO", "0.5"))
|
||||
except (TypeError, ValueError):
|
||||
logger.warning(
|
||||
"FB2_EMA_LO=%r is not a number — using tuned "
|
||||
"default 0.5", os.environ.get("FB2_EMA_LO"))
|
||||
_lo = 0.5
|
||||
if ema_dist_pct < _lo:
|
||||
report.warnings.append(f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk")
|
||||
penalties.append(("ema200_near", 5))
|
||||
|
||||
+201
-49
@@ -103,6 +103,12 @@ logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
|
||||
force=True)
|
||||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||||
# Never let DEBUG runs write credentials into the 30-day rotating log:
|
||||
# the websockets client logger emits the connect URI (Twelve Data puts
|
||||
# the API key in the query string) and frame contents (Tiingo's auth
|
||||
# frame carries the key) — 2026-08-04 audit, finding K47.
|
||||
logging.getLogger("websockets").setLevel(logging.INFO)
|
||||
logging.getLogger("websockets.client").setLevel(logging.INFO)
|
||||
logger = logging.getLogger("telegram_bot")
|
||||
|
||||
TOKEN = telegram_token()
|
||||
@@ -600,6 +606,45 @@ def _commit_alert(key: str, report: SignalReport) -> None:
|
||||
_last_alert_setup.pop(key, None)
|
||||
|
||||
|
||||
# Per-key locks serializing the _should_alert → send → _commit_alert
|
||||
# window. The auto-scan loop and a concurrent breach handler share the
|
||||
# "XAU/USD_15min" dedup key; without the lock both pass _should_alert
|
||||
# while the other's send is still awaiting Telegram, and the same signal
|
||||
# goes out twice (2026-08-04 audit H36). Per-key rather than one global
|
||||
# lock so a slow send on one timeframe cannot head-of-line-block another
|
||||
# timeframe's alert. Keys are symbol × TF, so the dict stays tiny.
|
||||
_alert_locks: Dict[str, asyncio.Lock] = {}
|
||||
|
||||
|
||||
def _alert_lock(key: str) -> asyncio.Lock:
|
||||
lock = _alert_locks.get(key)
|
||||
if lock is None:
|
||||
lock = _alert_locks[key] = asyncio.Lock()
|
||||
return lock
|
||||
|
||||
|
||||
async def _send_alert(bot, chat_id: int, text: str) -> bool:
|
||||
"""Markdown alert send with a plain-text fallback on parse failures,
|
||||
mirroring _safe_edit (2026-08-04 audit H35 — alert paths previously
|
||||
dropped the message on a BadRequest with only a warning logged).
|
||||
Returns True when a message reached the chat by either attempt."""
|
||||
try:
|
||||
await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
|
||||
return True
|
||||
except BadRequest as exc:
|
||||
logger.warning(f"Markdown alert failed for chat {chat_id} ({exc}); "
|
||||
f"retrying as plain text")
|
||||
try:
|
||||
await bot.send_message(chat_id, text)
|
||||
return True
|
||||
except Exception as exc2:
|
||||
logger.warning(f"Failed to send alert to chat {chat_id}: {exc2}")
|
||||
return False
|
||||
except Exception as exc:
|
||||
logger.warning(f"Failed to send alert to chat {chat_id}: {exc}")
|
||||
return False
|
||||
|
||||
|
||||
# ── Candle-boundary scheduler ────────────────────────────────────────
|
||||
|
||||
_autoscan_chats: Set[int] = set()
|
||||
@@ -628,8 +673,11 @@ def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
||||
elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
|
||||
interval_seconds = interval_minutes * 60
|
||||
remaining = interval_seconds - elapsed_seconds + buffer
|
||||
if remaining <= buffer:
|
||||
remaining += interval_seconds
|
||||
# NOTE: remaining > buffer always (elapsed < interval by construction);
|
||||
# a call at the exact boundary instant targets the NEXT boundary —
|
||||
# steady-state loops wake at boundary+buffer, so only a process
|
||||
# started in the boundary microsecond skips one scan (2026-08-04
|
||||
# audit I42; the old `remaining <= buffer` guard was unreachable).
|
||||
return remaining
|
||||
|
||||
|
||||
@@ -644,7 +692,13 @@ async def _scan_timeframe(
|
||||
reports: List[SignalReport] = []
|
||||
for i, result in enumerate(results):
|
||||
if isinstance(result, Exception):
|
||||
logger.exception(f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}")
|
||||
# gather(return_exceptions=True) means there is no active
|
||||
# exception here — logger.exception would log "NoneType: None"
|
||||
# instead of the real stack; pass the instance explicitly.
|
||||
logger.error(
|
||||
f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}",
|
||||
exc_info=result,
|
||||
)
|
||||
else:
|
||||
reports.append(result)
|
||||
|
||||
@@ -653,36 +707,51 @@ async def _scan_timeframe(
|
||||
|
||||
_update_level_cache(reports)
|
||||
|
||||
actionable = [
|
||||
candidates = [
|
||||
r for r in reports
|
||||
if r.final_signal != SignalStrength.NEUTRAL # same gate as breach path
|
||||
and _passes_quality(r.signal_quality)
|
||||
and _should_alert(f"{r.symbol}_{tf.value}", r)
|
||||
]
|
||||
sent_ok: Set[int] = set()
|
||||
for chat_id in _signal_targets():
|
||||
for i, report in enumerate(actionable):
|
||||
for report in candidates:
|
||||
key = f"{report.symbol}_{tf.value}"
|
||||
# The lock spans check → send → commit so a concurrent breach
|
||||
# handler on the same key cannot double-alert (H36).
|
||||
async with _alert_lock(key):
|
||||
if not _should_alert(key, report):
|
||||
continue
|
||||
detail = format_signal_message(report)
|
||||
try:
|
||||
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
|
||||
sent_ok.add(i)
|
||||
except Exception as exc:
|
||||
logger.warning(f"Failed to send signal for chat {chat_id}: {exc}")
|
||||
# Commit fingerprints only for alerts that reached at least one target;
|
||||
delivered = False
|
||||
for chat_id in _signal_targets():
|
||||
if await _send_alert(bot, chat_id, detail):
|
||||
delivered = True
|
||||
# Commit only when the alert reached at least one target;
|
||||
# a fully failed send stays eligible for the next scan.
|
||||
for i, report in enumerate(actionable):
|
||||
if i in sent_ok:
|
||||
_commit_alert(f"{report.symbol}_{tf.value}", report)
|
||||
if delivered:
|
||||
_commit_alert(key, report)
|
||||
|
||||
|
||||
# ── WebSocket level-breach monitor ──────────────────────────────────
|
||||
|
||||
_level_cache: Dict[str, Dict[str, float]] = {}
|
||||
_level_cache_ts: Dict[str, float] = {}
|
||||
# Level cache keyed symbol → timeframe → {level_name: price}, with a
|
||||
# matching per-TF write timestamp. Keyed per TF (2026-08-04 audit H34)
|
||||
# because every scan path writes here — the M5/M15/H1/H4 boundary loops,
|
||||
# the breach rescan, the manual Scan-All button — and a single flat dict
|
||||
# meant the last writer wiped the other timeframes' swing/EMA levels
|
||||
# within minutes. Breach checks scan the union of all fresh TF entries.
|
||||
_level_cache: Dict[str, Dict[str, Dict[str, float]]] = {}
|
||||
_level_cache_ts: Dict[str, Dict[str, float]] = {}
|
||||
_LEVEL_CACHE_TTL = 3600
|
||||
_last_prices: Dict[str, float] = {}
|
||||
_last_price_ts: Dict[str, float] = {}
|
||||
_scan_cooldowns: Dict[str, float] = {}
|
||||
# Breaches are edge-triggered (only the crossing tick fires), so one that
|
||||
# lands while the symbol is inside the rescan cooldown would otherwise be
|
||||
# dropped forever (H33b). Park it here; the tick loop fires it once the
|
||||
# cooldown expires. symbol → (price, level_name, recorded_at).
|
||||
_pending_breach: Dict[str, tuple] = {}
|
||||
# A failed breach rescan shrinks the just-set cooldown to this retry
|
||||
# window instead of silencing the symbol for the full cooldown (H33a).
|
||||
_BREACH_RETRY_SECONDS = 30
|
||||
|
||||
_cfg = PlatformConfig()
|
||||
_WS_COOLDOWN = _cfg.ws_cooldown_seconds
|
||||
@@ -691,6 +760,8 @@ _breach_tasks: Set[asyncio.Task] = set()
|
||||
|
||||
|
||||
def _update_level_cache(reports: List[SignalReport]) -> None:
|
||||
"""Each report updates only its own timeframe's slot — an M5 scan can
|
||||
no longer overwrite the H1/H4 swing levels (H34)."""
|
||||
now = time.time()
|
||||
for r in reports:
|
||||
if r.primary_analysis and r.primary_analysis.key_levels:
|
||||
@@ -699,26 +770,31 @@ def _update_level_cache(reports: List[SignalReport]) -> None:
|
||||
if k in _LEVEL_KEYS
|
||||
}
|
||||
if levels:
|
||||
_level_cache[r.symbol] = levels
|
||||
_level_cache_ts[r.symbol] = now
|
||||
tf = r.primary_analysis.timeframe
|
||||
_level_cache.setdefault(r.symbol, {})[tf] = levels
|
||||
_level_cache_ts.setdefault(r.symbol, {})[tf] = now
|
||||
|
||||
|
||||
def _check_level_breach(symbol: str, price: float) -> Optional[str]:
|
||||
levels = _level_cache.get(symbol)
|
||||
if not levels:
|
||||
return None
|
||||
|
||||
cache_age = time.time() - _level_cache_ts.get(symbol, 0)
|
||||
if cache_age > _LEVEL_CACHE_TTL:
|
||||
"""Scan the union of every fresh timeframe's levels; returns
|
||||
'<tf> <level_name>' (e.g. '15min swing_high_1') on a crossing so the
|
||||
alert label says which timeframe's level broke."""
|
||||
per_tf = _level_cache.get(symbol)
|
||||
if not per_tf:
|
||||
return None
|
||||
|
||||
last = _last_prices.get(symbol)
|
||||
if last is None:
|
||||
return None
|
||||
|
||||
now = time.time()
|
||||
ts_map = _level_cache_ts.get(symbol, {})
|
||||
for tf, levels in per_tf.items():
|
||||
if now - ts_map.get(tf, 0) > _LEVEL_CACHE_TTL:
|
||||
continue # stale TF entry — skip, fresher TFs still count
|
||||
for name, level in levels.items():
|
||||
if (last < level <= price) or (last > level >= price):
|
||||
return name
|
||||
return f"{tf} {name}"
|
||||
|
||||
return None
|
||||
|
||||
@@ -733,9 +809,17 @@ def _cleanup_cooldowns() -> None:
|
||||
async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) -> None:
|
||||
now = time.time()
|
||||
if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN:
|
||||
# Park the crossing instead of dropping it — price holding beyond
|
||||
# the level never re-fires the edge trigger, so without this the
|
||||
# breach would be lost forever (H33b). The tick loop fires it
|
||||
# once the cooldown expires; a newer crossing overwrites it.
|
||||
_pending_breach[pair_symbol] = (price, level_name, now)
|
||||
return
|
||||
|
||||
# Early-set on purpose: this is what stops concurrent breach tasks
|
||||
# from stacking rescans. The failure path below rolls it back.
|
||||
_scan_cooldowns[pair_symbol] = now
|
||||
_pending_breach.pop(pair_symbol, None) # this crossing is being handled
|
||||
display = PAIR_INFO.get(pair_symbol, {}).get("display", pair_symbol)
|
||||
label = level_name.replace("_", " ").title()
|
||||
logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15")
|
||||
@@ -745,29 +829,42 @@ async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) -
|
||||
_update_level_cache([report])
|
||||
except Exception:
|
||||
logger.exception(f"Breach scan failed for {display}")
|
||||
# Shrink the just-set cooldown to a short retry window so a
|
||||
# transient fetch error doesn't silence the symbol for the full
|
||||
# cooldown (H33a); the floor bounds the retry rate while the
|
||||
# provider stays down. Re-park the breach so the tick loop
|
||||
# retries it (setdefault: a newer crossing recorded meanwhile
|
||||
# wins over this stale one).
|
||||
retry = min(_BREACH_RETRY_SECONDS, _WS_COOLDOWN)
|
||||
_scan_cooldowns[pair_symbol] = now - _WS_COOLDOWN + retry
|
||||
_pending_breach.setdefault(pair_symbol, (price, level_name, now))
|
||||
return
|
||||
|
||||
# Alert only when the rescan produced an actionable, changed signal —
|
||||
# same gates as the auto-scanner, so a breach can't spam the channel.
|
||||
if (report.final_signal == SignalStrength.NEUTRAL
|
||||
or not _passes_quality(report.signal_quality)
|
||||
or not _should_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)):
|
||||
or not _passes_quality(report.signal_quality)):
|
||||
return
|
||||
|
||||
key = f"{pair_symbol}_{Timeframe.M15.value}"
|
||||
# Lock spans check → send → commit; without it this handler and an
|
||||
# in-flight M15 auto-scan both pass _should_alert and double-alert
|
||||
# the same signal (H36).
|
||||
async with _alert_lock(key):
|
||||
if not _should_alert(key, report):
|
||||
return
|
||||
|
||||
header = (
|
||||
f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed "
|
||||
f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n"
|
||||
)
|
||||
detail = header + format_signal_message(report)
|
||||
sent = False
|
||||
for chat_id in _signal_targets():
|
||||
detail = header + format_signal_message(report)
|
||||
try:
|
||||
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
|
||||
if await _send_alert(bot, chat_id, detail):
|
||||
sent = True
|
||||
except Exception as exc:
|
||||
logger.warning(f"Failed to send breach alert to chat {chat_id}: {exc}")
|
||||
if sent:
|
||||
_commit_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)
|
||||
_commit_alert(key, report)
|
||||
|
||||
|
||||
def _breach_task_done(task: asyncio.Task) -> None:
|
||||
@@ -799,13 +896,16 @@ async def _ws_monitor(bot) -> None:
|
||||
|
||||
backoff = 1
|
||||
while _feed_active():
|
||||
got_tick = False
|
||||
try:
|
||||
def _on_connect() -> None:
|
||||
nonlocal backoff
|
||||
async for pair_symbol, price in dp.iter_ticks(SCAN_PAIRS):
|
||||
if not got_tick:
|
||||
# Reset backoff only once the connection actually
|
||||
# produced data — resetting on connect let a
|
||||
# rejected key or accept-then-close server pin the
|
||||
# backoff at minimum forever (H38a/H38b).
|
||||
got_tick = True
|
||||
backoff = 1
|
||||
|
||||
async for pair_symbol, price in dp.iter_ticks(
|
||||
SCAN_PAIRS, on_connect=_on_connect):
|
||||
if not _feed_active():
|
||||
break
|
||||
|
||||
@@ -814,11 +914,32 @@ async def _ws_monitor(bot) -> None:
|
||||
_last_price_ts[pair_symbol] = time.time()
|
||||
|
||||
if breached:
|
||||
# A fresh crossing supersedes any parked one.
|
||||
_pending_breach.pop(pair_symbol, None)
|
||||
task = asyncio.create_task(
|
||||
_handle_breach(bot, pair_symbol, price, breached)
|
||||
)
|
||||
_breach_tasks.add(task)
|
||||
task.add_done_callback(_breach_task_done)
|
||||
elif pair_symbol in _pending_breach:
|
||||
pend = _pending_breach[pair_symbol]
|
||||
if time.time() - pend[2] > _LEVEL_CACHE_TTL:
|
||||
# Ancient parked breach (feed gap/weekend) — an
|
||||
# alert about an hours-old crossing would only
|
||||
# mislead; the level cache it came from is stale
|
||||
# anyway.
|
||||
_pending_breach.pop(pair_symbol)
|
||||
elif (time.time() - _scan_cooldowns.get(pair_symbol, 0)
|
||||
>= _WS_COOLDOWN):
|
||||
# Crossing parked during the cooldown: fire it now
|
||||
# that the window expired — the edge trigger alone
|
||||
# would have dropped it forever (H33b).
|
||||
_pending_breach.pop(pair_symbol)
|
||||
task = asyncio.create_task(
|
||||
_handle_breach(bot, pair_symbol, pend[0], pend[1])
|
||||
)
|
||||
_breach_tasks.add(task)
|
||||
task.add_done_callback(_breach_task_done)
|
||||
|
||||
_cleanup_cooldowns()
|
||||
|
||||
@@ -828,6 +949,16 @@ async def _ws_monitor(bot) -> None:
|
||||
logger.exception(f"WebSocket error, reconnecting in {backoff}s")
|
||||
await asyncio.sleep(backoff)
|
||||
backoff = min(backoff * 2, 60)
|
||||
else:
|
||||
# Clean server close: the async-for ended without raising.
|
||||
# This must back off and log like the error path — silently
|
||||
# reconnecting with zero delay hot-loops against a server
|
||||
# that accepts and immediately closes (H38a).
|
||||
if _feed_active():
|
||||
logger.warning(
|
||||
f"WebSocket closed cleanly, reconnecting in {backoff}s")
|
||||
await asyncio.sleep(backoff)
|
||||
backoff = min(backoff * 2, 60)
|
||||
|
||||
logger.info("WebSocket monitor stopped")
|
||||
|
||||
@@ -870,15 +1001,25 @@ _ws_task: Optional[asyncio.Task] = None
|
||||
|
||||
def _start_scheduler(bot) -> None:
|
||||
global _ws_task
|
||||
alive = [t for t in _autoscan_tasks if not t.done()]
|
||||
if not alive:
|
||||
_autoscan_tasks.clear()
|
||||
# Per-TF respawn (2026-08-04 audit H37): drop finished loop tasks and
|
||||
# recreate only the missing ones. The old all-or-nothing alive check
|
||||
# left a single dead TF loop unmonitored forever while its siblings
|
||||
# lived — and with the channel feed active, /autoscan never reaches
|
||||
# _stop_scheduler, so nothing short of a process restart recovered it.
|
||||
_autoscan_tasks[:] = [t for t in _autoscan_tasks if not t.done()]
|
||||
alive_names = {t.get_name() for t in _autoscan_tasks}
|
||||
started = []
|
||||
for tf, interval in _AUTOSCAN_LOOPS:
|
||||
name = f"candle_loop_{tf.name}"
|
||||
if name in alive_names:
|
||||
continue
|
||||
_autoscan_tasks.append(
|
||||
asyncio.create_task(_candle_loop(bot, tf, interval))
|
||||
asyncio.create_task(_candle_loop(bot, tf, interval), name=name)
|
||||
)
|
||||
started.append(tf.name)
|
||||
if started:
|
||||
logger.info("Candle-boundary scheduler started (%s)",
|
||||
"/".join(tf.name for tf, _ in _AUTOSCAN_LOOPS))
|
||||
"/".join(started))
|
||||
|
||||
if _ws_task is None or _ws_task.done():
|
||||
_ws_task = asyncio.create_task(_ws_monitor(bot))
|
||||
@@ -1006,9 +1147,16 @@ def _status_text() -> str:
|
||||
lines.append("\n_No live prices yet (WS not streaming)_")
|
||||
|
||||
cached_pairs = len(_level_cache)
|
||||
cached_levels = sum(len(v) for v in _level_cache.values())
|
||||
cached_levels = sum(
|
||||
len(levels) for per_tf in _level_cache.values()
|
||||
for levels in per_tf.values()
|
||||
)
|
||||
if cached_levels:
|
||||
oldest_age = time.time() - min(_level_cache_ts.values()) if _level_cache_ts else 0
|
||||
all_ts = [
|
||||
ts for per_tf in _level_cache_ts.values()
|
||||
for ts in per_tf.values()
|
||||
]
|
||||
oldest_age = time.time() - min(all_ts) if all_ts else 0
|
||||
lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs (oldest {oldest_age:.0f}s)")
|
||||
else:
|
||||
lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan")
|
||||
@@ -1311,7 +1459,11 @@ async def _post_init(app: Application) -> None:
|
||||
logger.error(f"Initial scan failed for {SCAN_PAIRS[i]}: {r}")
|
||||
if init_reports:
|
||||
_update_level_cache(init_reports)
|
||||
logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels")
|
||||
n_levels = sum(
|
||||
len(levels) for per_tf in _level_cache.values()
|
||||
for levels in per_tf.values()
|
||||
)
|
||||
logger.info(f"Level cache populated: {n_levels} levels")
|
||||
except Exception:
|
||||
logger.exception("Initial scan failed — WS will wait for first candle scan")
|
||||
|
||||
|
||||
+1170
-12
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user