Compare commits

..

3 Commits

Author SHA1 Message Date
Harwinder Sekhon 356a09e5ab Add commit-ladder campaign driver; ignore its 26MB results folder
Added: backtest/research/commit_ladder.py (recreates historical configs
via --set on the fixed harness; 2026-08-03 campaign confirmed HEAD is
the best config on all timeframes). Results stay local like all pinned
trade books.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Nsd8HcrV67GbCZXxZEn1Ma
2026-08-03 13:29:18 -04:00
Harwinder Sekhon 8521ab909a DeepSeek V4-Flash news enrichment: LLM sentiment, surprise direction, summary
Three display/warning-only upgrades to the news engine, all behind the
existing keyword path as fallback — signals are never gated on LLM
output (event_risk_caps_signals / sentiment_affects_signals stay False
per the 2026-07 audit; backtester use_news=False untouched):

- Headline sentiment: one batched call per 600s cache window (capped at
  25 new headlines, sha1-cached so repeats are never re-sent) overrides
  the keyword scores per currency with the gold-inverse doctrine encoded
  in the prompt.
- Release surprises: get_event_risk_level's description now says which
  way the print pulls GOLD ("— gold bearish: strong jobs lift yields"),
  fixing the documented numeric-only gap; risk level and windows
  unchanged.
- Market summary: 1-2 sentence macro blurb on SignalReport.news_summary,
  rendered in Telegram alerts (📰) and the CLI (🧠).

New llm_client.py: requests-only OpenAI-compatible client (no SDK), one
attempt per call, UTC-day request cap, singleton so the cap survives
per-scan NewsEngine reconstruction. Config: deepseek block in
config.json + DEEPSEEK_API_KEY env; absent key = fully inert.

Live-measured API behavior baked in: V4-Flash is a hybrid thinking model
— requests send thinking:disabled (reasoning otherwise blew the timeout
and could exhaust max_tokens before the answer); prompts are static
module constants so DeepSeek's prefix cache prices repeat calls at the
cache-hit rate; scores use compact [s, xau] pairs; timeout 12s.

Tests 273 → 286 (new Module 19 + deepseek config tests).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Nsd8HcrV67GbCZXxZEn1Ma
2026-08-03 13:25:02 -04:00
Harwinder Sekhon f0d76021d1 Tiingo WebSocket streaming: re-enable the level-breach monitor
The monitor has been dead since the Oanda feed switch (Oanda's provider
does not stream). New config.json scalar data_providers.streaming selects
a tick-stream provider independent of the candle feed; telegram_bot
resolves it via _get_streaming_dp() (separate instance, candle cache
stays on the active provider) and falls back to the active provider when
unset — behavior is byte-identical to before when the key is absent.

TiingoDataProvider grows the two sanctioned streaming overrides:
supports_streaming=True and iter_ticks() on wss://api.tiingo.com/fx
(mid price only — the no-spread rule extends to the stream). Two
deliberate divergences from the TD reference implementation, both
because Tiingo authenticates inside the socket: server "E" messages
raise TiingoDataError (a swallowed auth error would end the generator
cleanly and reconnect-hammer with zero backoff), and on_connect fires on
the first received message rather than after the subscribe send. Field
order of quote frames was live-verified: ask is data[7], not data[6] as
the docs table suggests (data[6] is askSize).

make_data_provider(config, name=None) now takes an explicit provider
name; keyless streaming providers disable the monitor with one warning
instead of a traceback-per-minute loop.

Tests 266 → 273 (WS parser, ticker round-trip, streaming config scalar,
explicit-name dispatch, streaming-dp resolution, monitor auto-disable).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Nsd8HcrV67GbCZXxZEn1Ma
2026-08-03 13:11:36 -04:00
14 changed files with 1243 additions and 23 deletions
+1
View File
@@ -35,3 +35,4 @@ backtest/trailgrid5m/
backtest/__pycache__/ backtest/__pycache__/
.env.* .env.*
logs/ logs/
backtest/research/commit_ladder_2026-08/
+5 -5
View File
@@ -11,7 +11,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest
## Commands ## Commands
```bash ```bash
python test_all.py # QA suite: 266 tests, 18 modules, no network (mocked) python test_all.py # QA suite: 286 tests, 19 modules, no network (mocked)
python main.py --pair XAUUSD --timeframe H1 --detail # CLI signal report (--json for full dump) 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 telegram_bot.py # the bot (refuses to boot without credentials)
python oanda_daemon.py # auto-execution daemon on Oanda practice account python oanda_daemon.py # auto-execution daemon on Oanda practice account
@@ -30,7 +30,7 @@ Python 3.12; a `venv/` exists in-repo. Dependencies in `requirements.txt` (numpy
## Configuration & credentials ## Configuration & credentials
Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_ACCOUNT_ID`/`OANDA_API_TOKEN`/`OANDA_BASE_URL`. Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `data_providers.streaming` (optional tick-stream provider for the level-breach monitor), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_ACCOUNT_ID`/`OANDA_API_TOKEN`/`OANDA_BASE_URL`, `DEEPSEEK_API_KEY` (news LLM; the `deepseek` config.json block holds model/enabled/timeout/daily cap).
## Architecture ## Architecture
@@ -49,12 +49,12 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz
### Layers ### Layers
- **`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` in `providers/__init__.py`; config.json `data_providers.active` selects the live feed. A test enforces that subclasses don't override shared base methods. Tiingo is candles-only — do not re-add spread display from its free FX feed (two-tier indicative quotes, removed 2026-07-23). - **`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. - **`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. `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]. - **`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). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`). - **`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), WebSocket level-breach monitor (auto-disabled when the active provider has `supports_streaming=False` — Oanda does not stream). Rotating log: `logs/telegram.log`, 3-h rotation × 240 backups. - **`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, trail ratchets at bar close exactly once per bar. Accepted divergence: one position total.
- **`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. - **`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.
+51 -2
View File
@@ -85,11 +85,15 @@ _VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
# AUTOSCAN_CONFIRMATIONS entries in telegram_bot.py). # AUTOSCAN_CONFIRMATIONS entries in telegram_bot.py).
_SCANNABLE_TFS = ("M5", "M15", "H1", "H4") _SCANNABLE_TFS = ("M5", "M15", "H1", "H4")
_KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level", "oanda"} _KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level",
"oanda", "deepseek"}
_KNOWN_OANDA = { _KNOWN_OANDA = {
"account_id", "api_token", "base_url", "lot_size", "account_id", "api_token", "base_url", "lot_size",
"quality_gate", "max_hold_bars", "quality_gate", "max_hold_bars",
} }
_KNOWN_DEEPSEEK = {
"api_key", "model", "enabled", "timeout_seconds", "daily_request_cap",
}
_KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"} _KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"}
_KNOWN_SCANNING = {"pairs", "timeframes"} _KNOWN_SCANNING = {"pairs", "timeframes"}
_KNOWN_PAIR_KEYS = {"blocked_hours_utc"} _KNOWN_PAIR_KEYS = {"blocked_hours_utc"}
@@ -106,12 +110,14 @@ class AppConfig:
telegram_channel_id: str = "" # "" = channel feed disabled telegram_channel_id: str = "" # "" = channel feed disabled
alert_min_quality: str = "C" alert_min_quality: str = "C"
active_provider: str = "twelve_data" active_provider: str = "twelve_data"
streaming_provider: Optional[str] = None # None = active provider
provider_settings: Dict[str, Dict[str, Any]] = field(default_factory=dict) provider_settings: Dict[str, Dict[str, Any]] = field(default_factory=dict)
scan_pairs: List[str] = field(default_factory=lambda: [XAUUSD]) scan_pairs: List[str] = field(default_factory=lambda: [XAUUSD])
scan_timeframes: List[Timeframe] = field(default_factory=lambda: [ scan_timeframes: List[Timeframe] = field(default_factory=lambda: [
Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]) Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4])
pair_overrides: Dict[str, Dict[str, Any]] = field(default_factory=dict) pair_overrides: Dict[str, Dict[str, Any]] = field(default_factory=dict)
oanda_settings: Dict[str, Any] = field(default_factory=dict) oanda_settings: Dict[str, Any] = field(default_factory=dict)
deepseek_settings: Dict[str, Any] = field(default_factory=dict)
log_level: str = "INFO" log_level: str = "INFO"
@@ -162,7 +168,8 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig:
app.active_provider = str(dp.get("active", app.active_provider)) app.active_provider = str(dp.get("active", app.active_provider))
app.provider_settings = {} app.provider_settings = {}
for name, block in dp.items(): for name, block in dp.items():
if name == "active": # "active" and "streaming" are reserved scalars, not blocks.
if name in ("active", "streaming"):
continue continue
if not isinstance(block, dict): if not isinstance(block, dict):
raise AppConfigError( raise AppConfigError(
@@ -172,6 +179,17 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig:
raise AppConfigError( raise AppConfigError(
f'config.json: data_providers.active "{app.active_provider}" ' f'config.json: data_providers.active "{app.active_provider}" '
"has no matching provider block") "has no matching provider block")
streaming = dp.get("streaming")
if streaming is not None:
if not isinstance(streaming, str):
raise AppConfigError(
'config.json: data_providers.streaming must be a string '
"(a provider name)")
if streaming not in app.provider_settings:
raise AppConfigError(
f'config.json: data_providers.streaming "{streaming}" '
"has no matching provider block")
app.streaming_provider = streaming
scanning = raw.get("scanning", {}) scanning = raw.get("scanning", {})
_check_keys(scanning, _KNOWN_SCANNING, '"scanning"') _check_keys(scanning, _KNOWN_SCANNING, '"scanning"')
@@ -235,6 +253,29 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig:
"config.json: oanda.max_hold_bars must be a positive integer") "config.json: oanda.max_hold_bars must be a positive integer")
app.oanda_settings = dict(oanda) app.oanda_settings = dict(oanda)
deepseek = raw.get("deepseek", {})
_check_keys(deepseek, _KNOWN_DEEPSEEK, '"deepseek"')
if "enabled" in deepseek and not isinstance(deepseek["enabled"], bool):
raise AppConfigError(
"config.json: deepseek.enabled must be true or false")
if "timeout_seconds" in deepseek:
ts = deepseek["timeout_seconds"]
if isinstance(ts, bool) or not isinstance(ts, (int, float)) or ts <= 0:
raise AppConfigError(
"config.json: deepseek.timeout_seconds must be a positive "
"number")
if "daily_request_cap" in deepseek:
cap = deepseek["daily_request_cap"]
if isinstance(cap, bool) or not isinstance(cap, int) or cap < 1:
raise AppConfigError(
"config.json: deepseek.daily_request_cap must be a positive "
"integer")
if "model" in deepseek:
if not isinstance(deepseek["model"], str) or not deepseek["model"]:
raise AppConfigError(
"config.json: deepseek.model must be a non-empty string")
app.deepseek_settings = dict(deepseek)
level = raw.get("log_level", app.log_level) level = raw.get("log_level", app.log_level)
if level not in _VALID_LOG_LEVELS: if level not in _VALID_LOG_LEVELS:
raise AppConfigError( raise AppConfigError(
@@ -326,6 +367,14 @@ def oanda_base_url(app: Optional[AppConfig] = None) -> str:
app.oanda_settings.get("base_url", "https://api-fxpractice.oanda.com")) app.oanda_settings.get("base_url", "https://api-fxpractice.oanda.com"))
# ── DeepSeek resolution helper (env → config.json → "") ──────────────
def deepseek_api_key(app: Optional[AppConfig] = None) -> str:
app = app or get_app_config()
return os.environ.get("DEEPSEEK_API_KEY") or str(
app.deepseek_settings.get("api_key", ""))
# ── PlatformConfig overlay ──────────────────────────────────────────── # ── PlatformConfig overlay ────────────────────────────────────────────
def apply_pair_overrides( def apply_pair_overrides(
+178
View File
@@ -0,0 +1,178 @@
"""Commit-ladder campaign driver (2026-08).
Question: of all signal-code states shipped in the last two months, which
configuration is the most accurate? Historical in-era backtest numbers are
not comparable (the backtester's realism model changed on 07-08/07-16 and
the data feed switched to Oanda on 07-28), so every historical
configuration is recreated on TODAY's harness via --set overrides and
replayed on the same pinned Oanda bars. Bug-fix commits stay on; the
ladder compares shipped tuning decisions only.
Usage:
python backtest/research/commit_ladder.py # full matrix
python backtest/research/commit_ladder.py --tf 5min # one timeframe
python backtest/research/commit_ladder.py --windows 12mo,oos
Resumable: a run whose CSV already exists in the output dir is skipped.
Outputs: backtest/research/commit_ladder_2026-08/<variant>_<window>.csv
+ .log per run, summary.csv, and a printed ranking table.
"""
from __future__ import annotations
import argparse
import subprocess
import sys
from pathlib import Path
import pandas as pd
ROOT = Path(__file__).resolve().parents[2]
BT = ROOT / "backtest" / "gpu_backtest.py"
OUT = ROOT / "backtest" / "research" / "commit_ladder_2026-08"
WINDOWS = {
"12mo": ("2025-07-30", "2026-07-30"),
"oos": ("2024-06-01", "2025-03-30"),
"5yr": ("2021-07-30", "2026-07-30"),
}
SPLIT_12MO = pd.Timestamp("2026-01-29", tz="UTC") # book_diff.py's split
# (variant, "commit date what-it-represents", [--set overrides])
VARIANTS: dict[str, list[tuple[str, str, list[str]]]] = {
"5min": [
("M5-A", "dad2432 07-09 gates+trailing+pullback",
["xau_strategy.blocked_hours_utc=[0,1,2,10,11,13,16,21]",
"xau_strategy.cp_enabled=false",
"signals.trail_dist_r=0.75",
"xau_strategy.m5_min_abs_score=30",
"xau_strategy.m5_min_confirmation_score=10"]),
("M5-B", "f3ba4d5 07-11 +hour-9 block",
["xau_strategy.blocked_hours_utc=[0,1,2,10,11,13,16,21,9]",
"xau_strategy.cp_enabled=false",
"signals.trail_dist_r=0.75",
"xau_strategy.m5_min_abs_score=30",
"xau_strategy.m5_min_confirmation_score=10"]),
("M5-C", "de6424e 07-16 +chart patterns",
["xau_strategy.blocked_hours_utc=[0,1,2,10,11,13,16,21,9]",
"signals.trail_dist_r=0.75",
"xau_strategy.m5_min_abs_score=30",
"xau_strategy.m5_min_confirmation_score=10"]),
("M5-D", "3ee79b5 07-21 +trail dist 0.85",
["xau_strategy.blocked_hours_utc=[0,1,2,10,11,13,16,21,9]",
"xau_strategy.m5_min_abs_score=30",
"xau_strategy.m5_min_confirmation_score=10"]),
("M5-E", "0396970 07-30 relaxation (HEAD)", []),
],
"15min": [
("M15-A", "987ec5b 07-09 pre-campaigns",
["signals.tp_floor_rr=0.5",
"xau_strategy.blocked_regimes=[]",
"xau_strategy.sell_blocked_hours_utc=[]",
"signals.trail_arm_r=0.60"]),
("M15-B", "f3ba4d5/f88e39f 07-11 1R floor + regime block",
["xau_strategy.sell_blocked_hours_utc=[]",
"signals.trail_arm_r=0.60"]),
("M15-C", "7d3fd95 07-16 +SELL hour 15/17 block",
["signals.trail_arm_r=0.60"]),
("M15-D", "bf8e5fb 07-21 trail arm 0.70 (HEAD)", []),
],
"1h": [
("H1-A", "pre-d95d31a 07-08 TP-tightening model",
["signals.min_rr_ratio=0.4",
"signals.tp_tightening_factor=0.8"]),
("H1-B", "HEAD", []),
],
}
def run_one(tf: str, name: str, overrides: list[str], wkey: str,
spread: str | None = None) -> Path:
suffix = f"_s{spread}" if spread else ""
csv_path = OUT / f"{name}_{wkey}{suffix}.csv"
log_path = OUT / f"{name}_{wkey}{suffix}.log"
if csv_path.exists():
print(f" [skip] {csv_path.name} exists")
return csv_path
start, end = WINDOWS[wkey]
cmd = [sys.executable, str(BT), "--pair", "XAUUSD",
"--primary-tf", tf, "--start", start, "--end", end,
"--export-csv", str(csv_path)]
if spread:
cmd += ["--spread", spread]
for ov in overrides:
cmd += ["--set", ov]
print(f" [run ] {csv_path.name} ...", flush=True)
with open(log_path, "w") as lf:
rc = subprocess.run(cmd, stdout=lf, stderr=subprocess.STDOUT,
cwd=ROOT).returncode
if rc != 0:
raise SystemExit(f"backtest failed (rc={rc}) — see {log_path}")
log = log_path.read_text()
n_ov = log.count("--set override")
if n_ov != len(overrides):
raise SystemExit(
f"{name}_{wkey}: expected {len(overrides)} --set overrides, "
f"log shows {n_ov} — see {log_path}")
if "Loading cached" not in log and wkey != "sanity":
print(f" [warn] {log_path.name}: no 'Loading cached' line — "
f"data was fetched live for at least one timeframe")
return csv_path
def metrics(csv_path: Path, wkey: str) -> dict:
df = pd.read_csv(csv_path, parse_dates=["entry_time"])
if df.empty:
return {"trades": 0, "total_r": 0.0, "wr": 0.0, "pf": 0.0,
"max_dd": 0.0, "h1_r": 0.0, "h2_r": 0.0, "years_pos": ""}
if df["entry_time"].dt.tz is None:
df["entry_time"] = df["entry_time"].dt.tz_localize("UTC")
r = df["r_multiple"]
pos, neg = r[r > 0].sum(), r[r < 0].sum()
cum = r.cumsum()
dd = (pd.concat([pd.Series([0.0]), cum]).cummax()[1:] - cum).max()
out = {
"trades": len(df),
"total_r": round(r.sum(), 1),
"wr": round((r > 0).mean() * 100, 1),
"pf": round(pos / abs(neg), 2) if neg else float("inf"),
"max_dd": round(dd, 1),
"h1_r": "", "h2_r": "", "years_pos": "",
}
if wkey == "12mo":
h1 = r[df["entry_time"] < SPLIT_12MO].sum()
out["h1_r"] = round(h1, 1)
out["h2_r"] = round(r.sum() - h1, 1)
if wkey == "5yr":
by_year = r.groupby(df["entry_time"].dt.year).sum()
out["years_pos"] = f"{(by_year > 0).sum()}/{len(by_year)}"
return out
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--tf", choices=list(VARIANTS), default=None)
ap.add_argument("--windows", default="12mo,oos,5yr")
args = ap.parse_args()
OUT.mkdir(parents=True, exist_ok=True)
wkeys = [w.strip() for w in args.windows.split(",")]
tfs = [args.tf] if args.tf else list(VARIANTS)
rows = []
for tf in tfs:
for name, label, overrides in VARIANTS[tf]:
for wkey in wkeys:
csv_path = run_one(tf, name, overrides, wkey)
m = metrics(csv_path, wkey)
rows.append({"tf": tf, "variant": name, "represents": label,
"window": wkey, **m})
print(f" {name} {wkey}: {m['trades']} trades, "
f"{m['total_r']}R, PF {m['pf']}, WR {m['wr']}%")
summary = pd.DataFrame(rows)
summary.to_csv(OUT / "summary.csv", index=False)
print("\n=== SUMMARY (also in summary.csv) ===")
print(summary.to_string(index=False))
if __name__ == "__main__":
main()
+8
View File
@@ -7,6 +7,7 @@
}, },
"data_providers": { "data_providers": {
"active": "oanda", "active": "oanda",
"streaming": "tiingo",
"twelve_data": { "twelve_data": {
"api_key": "REPLACE_WITH_YOUR_TWELVE_DATA_API_KEY" "api_key": "REPLACE_WITH_YOUR_TWELVE_DATA_API_KEY"
}, },
@@ -32,5 +33,12 @@
"quality_gate": "C", "quality_gate": "C",
"max_hold_bars": 300 "max_hold_bars": 300
}, },
"deepseek": {
"api_key": "REPLACE_WITH_YOUR_DEEPSEEK_API_KEY_OR_EMPTY",
"model": "deepseek-v4-flash",
"enabled": true,
"timeout_seconds": 12,
"daily_request_cap": 500
},
"log_level": "INFO" "log_level": "INFO"
} }
+1
View File
@@ -69,6 +69,7 @@
<tr><td><b>Score / confidence</b></td><td>The regime-weighted composite of all nine indicators. Bigger absolute score = stronger consensus.</td></tr> <tr><td><b>Score / confidence</b></td><td>The regime-weighted composite of all nine indicators. Bigger absolute score = stronger consensus.</td></tr>
<tr><td><b>MTF aligned</b></td><td>The two confirmation timeframes agree with the trade direction. Signals without alignment are filtered out on M5.</td></tr> <tr><td><b>MTF aligned</b></td><td>The two confirmation timeframes agree with the trade direction. Signals without alignment are filtered out on M5.</td></tr>
<tr><td><b>Warnings ⚠️</b></td><td>Context, not veto. Backtesting shows warned trades perform about the same — strong momentum often triggers warnings.</td></tr> <tr><td><b>Warnings ⚠️</b></td><td>Context, not veto. Backtesting shows warned trades perform about the same — strong momentum often triggers warnings.</td></tr>
<tr><td><b>Market context 📰</b></td><td>An AI-written one-liner summarizing today's macro backdrop (headlines + high-impact releases) and its likely pull on gold. Informational only — it never blocks or grades a signal.</td></tr>
<tr><td><b>SL / TP1 / RR</b></td><td>Stop loss sits behind the nearest swing level with an ATR buffer. TP1 is the nearest opposing level. <b>1R</b> = the entry-to-SL distance = the amount you risk.</td></tr> <tr><td><b>SL / TP1 / RR</b></td><td>Stop loss sits behind the nearest swing level with an ATR buffer. TP1 is the nearest opposing level. <b>1R</b> = the entry-to-SL distance = the amount you risk.</td></tr>
<tr><td><b>Trail plan 🔄</b></td><td>The real exit. Once price moves +0.6R in your favor (+0.8R for pullback setups), move the stop to trail 0.75R behind the best price and update it at each candle close. <b>TP1 is advisory — let winners run.</b></td></tr> <tr><td><b>Trail plan 🔄</b></td><td>The real exit. Once price moves +0.6R in your favor (+0.8R for pullback setups), move the stop to trail 0.75R behind the best price and update it at each candle close. <b>TP1 is advisory — let winners run.</b></td></tr>
</table> </table>
+185
View File
@@ -0,0 +1,185 @@
"""
ForexBot2 — DeepSeek LLM client
================================
Thin requests-based client for DeepSeek's OpenAI-compatible chat API,
used by `news_engine` for display/warning-only enrichment (headline
sentiment, event surprise direction, market summary). Signals are NEVER
gated on LLM output — every caller has a non-LLM fallback and treats a
`None` return as "use the fallback".
Design constraints:
- One attempt per call, no retries: news runs inside scan cycles; the
news engine's 600 s context cache is the natural retry cadence.
- A UTC-day request cap bounds spend even if a bug causes call storms.
- Cost: DeepSeek caches identical prompt PREFIXES server-side (cache-hit
input is ~50x cheaper), so callers pass a STATIC system prompt (no
timestamps or interpolation) and put all variable content in the user
payload.
This is a module-level singleton (`get_llm_client`) because the daily
cap counter must survive the per-scan reconstruction of `NewsEngine`.
"""
import datetime as _dt
import json
import logging
import threading
from typing import Any, Dict, Optional
import requests
from app_config import deepseek_api_key, get_app_config
logger = logging.getLogger(__name__)
DEEPSEEK_URL = "https://api.deepseek.com/chat/completions"
DEFAULT_MODEL = "deepseek-v4-flash"
# 8s timed out on a measured cold-prefix-cache headline batch (2026-08-03);
# warm-cache calls come back well under that. 12s buys cold-start headroom
# while still bounding worst-case scan latency (≤3 calls per cache window).
DEFAULT_TIMEOUT = 12.0
DEFAULT_DAILY_CAP = 500
class DeepSeekClient:
"""JSON-in/JSON-out chat wrapper. All failures return None."""
def __init__(
self,
api_key: str,
model: str = DEFAULT_MODEL,
timeout: float = DEFAULT_TIMEOUT,
daily_cap: int = DEFAULT_DAILY_CAP,
enabled: bool = True,
):
self._api_key = api_key or ""
self._model = model
self._timeout = timeout
self._daily_cap = daily_cap
self._enabled = enabled
self._session = requests.Session()
self._session.headers.update({
"Content-Type": "application/json",
"Authorization": f"Bearer {self._api_key}",
})
self._lock = threading.Lock()
self._count_day: Optional[_dt.date] = None
self._count = 0
self._warned: set = set()
@property
def available(self) -> bool:
return self._enabled and bool(self._api_key)
def _warn_once(self, kind: str, msg: str, *args: Any) -> None:
if kind in self._warned:
logger.debug(msg, *args)
else:
self._warned.add(kind)
logger.warning(msg, *args)
def _under_cap(self) -> bool:
"""Count a request against the UTC-day cap; False = don't call."""
today = _dt.datetime.now(_dt.timezone.utc).date()
with self._lock:
if self._count_day != today:
self._count_day = today
self._count = 0
self._warned.discard("cap")
if self._count >= self._daily_cap:
self._warn_once(
"cap", "DeepSeek daily request cap (%d) exhausted — "
"falling back to keyword analysis until UTC midnight",
self._daily_cap)
return False
self._count += 1
return True
def chat_json(self, system_prompt: str,
user_payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""One chat call; returns the model's JSON object or None."""
if not self.available or not self._under_cap():
return None
body = {
"model": self._model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user",
"content": json.dumps(user_payload, sort_keys=True)},
],
"response_format": {"type": "json_object"},
"temperature": 0,
# V4-Flash is a hybrid thinking model; without this flag it
# burns hundreds of reasoning tokens per call — measured
# 2026-08-03: a 25-headline batch blew straight through the
# timeout AND could exhaust max_tokens before the answer,
# returning empty content. The doctrine + few-shot examples
# in the system prompts carry the quality instead.
"thinking": {"type": "disabled"},
# Must fit the largest reply: a full sentiment batch is
# ~10 tokens per headline; 1000 truncated mid-JSON on a real
# RSS cycle (2026-08-03).
"max_tokens": 2000,
}
try:
resp = self._session.post(
DEEPSEEK_URL, json=body, timeout=self._timeout)
except requests.RequestException as e:
self._warn_once("network", "DeepSeek call failed: %s", e)
return None
if resp.status_code != 200:
self._warn_once(
f"http_{resp.status_code}",
"DeepSeek HTTP %s: %s", resp.status_code, resp.text[:200])
return None
try:
parsed = self._parse_response(resp.json())
except ValueError:
parsed = None
if parsed is None:
self._warn_once("parse", "DeepSeek response was not the "
"expected JSON object — ignoring")
return parsed
@staticmethod
def _parse_response(body: Any) -> Optional[Dict[str, Any]]:
"""Extract the assistant's JSON object from a chat completion."""
try:
content = body["choices"][0]["message"]["content"]
except (KeyError, IndexError, TypeError):
return None
try:
obj = json.loads(content)
except (TypeError, ValueError):
return None
return obj if isinstance(obj, dict) else None
# ── Process-wide singleton ────────────────────────────────────────────
_client: Optional[DeepSeekClient] = None
_client_lock = threading.Lock()
def get_llm_client() -> DeepSeekClient:
global _client
with _client_lock:
if _client is None:
settings = get_app_config().deepseek_settings
_client = DeepSeekClient(
api_key=deepseek_api_key(),
model=str(settings.get("model", DEFAULT_MODEL)),
timeout=float(settings.get("timeout_seconds",
DEFAULT_TIMEOUT)),
daily_cap=int(settings.get("daily_request_cap",
DEFAULT_DAILY_CAP)),
enabled=settings.get("enabled", True) is not False,
)
return _client
def reset_llm_client() -> None:
"""Drop the singleton (tests / config reloads)."""
global _client
with _client_lock:
_client = None
+3
View File
@@ -161,6 +161,8 @@ def print_signal_report(report: SignalReport, detail: bool = False):
print(f" {Colors.YELLOW}📅{Colors.RESET} {report.event_description}") print(f" {Colors.YELLOW}📅{Colors.RESET} {report.event_description}")
if report.news_description and report.news_description != "Neutral news sentiment": if report.news_description and report.news_description != "Neutral news sentiment":
print(f" {Colors.MAGENTA}📰{Colors.RESET} {report.news_description}") print(f" {Colors.MAGENTA}📰{Colors.RESET} {report.news_description}")
if report.news_summary:
print(f" {Colors.MAGENTA}🧠{Colors.RESET} {report.news_summary}")
# ── 6. DETAIL (only with --detail) ───────────────────────────── # ── 6. DETAIL (only with --detail) ─────────────────────────────
if detail and report.primary_analysis: if detail and report.primary_analysis:
@@ -206,6 +208,7 @@ def report_to_dict(report: SignalReport) -> dict:
"event_description": report.event_description, "event_description": report.event_description,
"news_sentiment": round(report.news_sentiment, 3), "news_sentiment": round(report.news_sentiment, 3),
"news_description": report.news_description, "news_description": report.news_description,
"news_summary": report.news_summary,
"bullish_reasons": report.bullish_reasons, "bullish_reasons": report.bullish_reasons,
"bearish_reasons": report.bearish_reasons, "bearish_reasons": report.bearish_reasons,
"warnings": report.warnings, "warnings": report.warnings,
+208 -1
View File
@@ -8,6 +8,7 @@ Key principle: High-impact news events override technical signals.
A perfect technical setup can be destroyed by an unexpected NFP print. A perfect technical setup can be destroyed by an unexpected NFP print.
""" """
import hashlib
import logging import logging
import re import re
import threading import threading
@@ -21,10 +22,62 @@ from typing import Dict, List, Optional, Tuple
import requests import requests
from config import NewsConfig from config import NewsConfig
from llm_client import get_llm_client
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# ── DeepSeek prompts ─────────────────────────────────────────────────
# STATIC constants on purpose: DeepSeek prices a repeated byte-identical
# prompt prefix at the cache-hit rate (~50x cheaper), so nothing here may
# be interpolated or timestamped. All variable content travels in the
# user payload. LLM output is display/warning-only — it never gates
# signals (see NewsConfig.sentiment_affects_signals).
_LLM_SENTIMENT_SYSTEM = """\
You are a forex/gold news analyst. You receive JSON:
{"headlines": ["...", ...]}
For EACH headline, in the same order, score:
- "s": overall market sentiment of the headline, -1.0 (very bearish)
to 1.0 (very bullish), 0.0 if irrelevant or neutral.
- "xau": the effect on the GOLD (XAU/USD) price, -1.0 (gold down)
to 1.0 (gold up), 0.0 if irrelevant.
Gold doctrine: hawkish Fed / rate hikes / rising real yields / strong
dollar are gold-bearish; easing / rate cuts / safe-haven flows /
geopolitical stress / weak dollar are gold-bullish.
Handle negation correctly ("no rate hike" is not hawkish).
Reply with ONLY JSON: {"scores": [[s, xau], ...]} — exactly one
[s, xau] pair per input headline, in the same order, nothing else.
Examples:
"Fed signals faster rate hikes as inflation stays hot" -> [0.4, -0.7]
"Gold surges to record high on safe-haven demand" -> [0.8, 0.9]
"Dollar slips after soft jobs report; Treasury yields fall" -> [-0.4, 0.6]
"""
_LLM_EVENT_SYSTEM = """\
You are a forex/gold analyst. You receive one released economic event as
JSON: {"event", "currency", "actual", "forecast", "previous"}.
Decide the likely effect of this RELEASE (actual vs forecast) on the
GOLD (XAU/USD) price. A stronger-than-forecast USD economy or a hawkish
implication is usually gold-bearish; weaker data or a dovish implication
is usually gold-bullish. Mind inverted metrics: higher unemployment or
jobless claims = weaker economy.
Reply with ONLY JSON:
{"gold_direction": "bullish"|"bearish"|"neutral", "reason": "<max 10 words>"}
"""
_LLM_SUMMARY_SYSTEM = """\
You are a forex/gold market-context writer. You receive JSON with recent
"headlines" and today's high-impact "events" (released: true means the
number is out). Write a 1-2 sentence plain-English summary of the
current macro backdrop for a GOLD (XAU/USD) trader: the dominant theme
and its likely pull on gold. No advice, no hedging boilerplate, max 45
words.
Reply with ONLY JSON: {"summary": "..."}
"""
_RELEASE_VALUE_RE = re.compile(r"(-?\d+(?:\.\d+)?)\s*([kmbt])?", re.IGNORECASE) _RELEASE_VALUE_RE = re.compile(r"(-?\d+(?:\.\d+)?)\s*([kmbt])?", re.IGNORECASE)
_RELEASE_MULTIPLIERS = {"k": 1e3, "m": 1e6, "b": 1e9, "t": 1e12} _RELEASE_MULTIPLIERS = {"k": 1e3, "m": 1e6, "b": 1e9, "t": 1e12}
@@ -105,6 +158,7 @@ class FundamentalContext:
news_headlines: List[NewsSentiment] = field(default_factory=list) news_headlines: List[NewsSentiment] = field(default_factory=list)
event_impact_on_pairs: Dict[str, str] = field(default_factory=dict) event_impact_on_pairs: Dict[str, str] = field(default_factory=dict)
calendar_available: bool = False # True if live calendar data was fetched calendar_available: bool = False # True if live calendar data was fetched
llm_summary: str = "" # DeepSeek market-context blurb (display-only)
class NewsEngine: class NewsEngine:
@@ -185,6 +239,17 @@ class NewsEngine:
_headlines_cache: Tuple[float, List[NewsSentiment]] = (0.0, []) _headlines_cache: Tuple[float, List[NewsSentiment]] = (0.0, [])
_EMPTY_RESULT_TTL = 60.0 # retry sooner after a failed/empty fetch _EMPTY_RESULT_TTL = 60.0 # retry sooner after a failed/empty fetch
# DeepSeek result caches. Headlines and releases are immutable, so a
# scored one never needs re-scoring; bounded by dropping the oldest
# insertion. _llm_sent_cache is only touched under _cache_lock (its
# writer runs inside _cached_fetch); the other two take the lock
# around cache access but keep HTTP calls outside it.
_llm_sent_cache: Dict[str, Tuple[float, float]] = {} # sha1 → (s, xau)
_llm_event_cache: Dict[Tuple[str, str, str], str] = {}
_llm_summary_cache: Tuple[float, str] = (0.0, "")
_LLM_SENT_CACHE_MAX = 400
_LLM_EVENT_CACHE_MAX = 100
def __init__(self, config: NewsConfig): def __init__(self, config: NewsConfig):
self.config = config self.config = config
self._session = requests.Session() self._session = requests.Session()
@@ -264,6 +329,12 @@ class NewsEngine:
except Exception as e: except Exception as e:
logger.warning(f"News sentiment fetch failed: {e}") logger.warning(f"News sentiment fetch failed: {e}")
# 3. LLM market-context summary (display-only)
try:
context.llm_summary = self._llm_market_summary(context)
except Exception as e:
logger.debug(f"LLM market summary failed: {e}")
return context return context
@staticmethod @staticmethod
@@ -311,10 +382,17 @@ class NewsEngine:
) )
if event.minutes_until > -window: if event.minutes_until > -window:
if surprised: if surprised:
return "caution", ( desc = (
f"{event.currency} {event.event} released — {surprise} surprise " f"{event.currency} {event.event} released — {surprise} surprise "
f"(actual {event.actual} vs forecast {event.forecast})" f"(actual {event.actual} vs forecast {event.forecast})"
) )
# LLM enrichment touches the DESCRIPTION only —
# `surprise` is numeric-direction and does not know
# whether a beat is gold-bullish; the LLM does.
gold_dir = self._llm_event_direction(event)
if gold_dir:
desc += f" — gold {gold_dir}"
return "caution", desc
return "caution", f"{event.currency} {event.event} just released" return "caution", f"{event.currency} {event.event} just released"
return "clear", "No significant events" return "clear", "No significant events"
@@ -596,6 +674,11 @@ class NewsEngine:
except Exception as e: except Exception as e:
logger.debug(f"Failed to fetch RSS from {feed_url}: {e}") logger.debug(f"Failed to fetch RSS from {feed_url}: {e}")
try:
self._apply_llm_sentiment(headlines)
except Exception as e:
logger.debug(f"LLM sentiment failed, keyword scores stand: {e}")
return headlines return headlines
def _analyze_headline_sentiment( def _analyze_headline_sentiment(
@@ -637,6 +720,130 @@ class NewsEngine:
return (bullish_score - bearish_score) / total return (bullish_score - bearish_score) / total
# ── DeepSeek enrichment (display/warning-only, keyword fallback) ──
def _llm_headline_scores(
self, texts: List[str]
) -> Optional[List[Tuple[float, float]]]:
"""One batched DeepSeek call → [(overall, xau), ...] aligned with
`texts`, or None on any failure/mismatch (caller keeps keyword
scores). Tests stub this method directly."""
result = get_llm_client().chat_json(
_LLM_SENTIMENT_SYSTEM, {"headlines": texts})
if not result:
return None
scores = result.get("scores")
if not isinstance(scores, list) or len(scores) != len(texts):
return None
out = []
for item in scores:
try:
s = max(-1.0, min(1.0, float(item[0])))
x = max(-1.0, min(1.0, float(item[1])))
except (TypeError, KeyError, ValueError, IndexError):
return None
out.append((s, x))
return out
def _apply_llm_sentiment(self, headlines: List[NewsSentiment]) -> None:
"""Overwrite keyword sentiment on `headlines` with LLM scores.
Runs inside _cached_fetch, i.e. UNDER _cache_lock — cache access
is direct and the (rare) HTTP call blocks sibling scans at most
once per cache window, matching the existing shared-fetch design.
Only headlines missing from the cache are sent; a failed batch
changes nothing.
"""
if not headlines or not get_llm_client().available:
return
cache = NewsEngine._llm_sent_cache
keyed = [(hashlib.sha1(h.headline.encode()).hexdigest(), h)
for h in headlines]
# Cap the batch so the reply fits the client's max_tokens — the
# overflow stays keyword-scored this cycle and, still uncached,
# goes out with the next window's batch.
new = [(k, h) for k, h in keyed if k not in cache][:25]
if new:
scores = self._llm_headline_scores([h.headline for _, h in new])
if scores is not None:
for (k, _), sc in zip(new, scores):
cache[k] = sc
while len(cache) > self._LLM_SENT_CACHE_MAX:
cache.pop(next(iter(cache)))
for k, h in keyed:
sc = cache.get(k)
if sc is None:
continue
overall, xau = sc
h.sentiment_score = overall
h.sentiment_by_currency = {
cur: (xau if cur == "XAU" else overall)
for cur in h.currencies_mentioned
}
def _llm_event_direction(self, event: EconomicEvent) -> Optional[str]:
"""Gold direction for a released surprise, e.g. "bearish: strong
jobs lift yields". None → caller keeps the numeric description.
Cached per release; failures are NOT cached (retry next scan)."""
key = (event.event, event.currency, str(event.actual or ""))
with NewsEngine._cache_lock:
if key in NewsEngine._llm_event_cache:
return NewsEngine._llm_event_cache[key] or None
client = get_llm_client()
if not client.available:
return None
result = client.chat_json(_LLM_EVENT_SYSTEM, {
"event": event.event,
"currency": event.currency,
"actual": event.actual,
"forecast": event.forecast,
"previous": event.previous,
})
if not result:
return None
direction = result.get("gold_direction")
if direction not in ("bullish", "bearish", "neutral"):
return None
reason = str(result.get("reason", "")).strip()[:80]
text = "" if direction == "neutral" else (
f"{direction}: {reason}" if reason else direction)
with NewsEngine._cache_lock:
NewsEngine._llm_event_cache[key] = text
while len(NewsEngine._llm_event_cache) > self._LLM_EVENT_CACHE_MAX:
NewsEngine._llm_event_cache.pop(
next(iter(NewsEngine._llm_event_cache)))
return text or None
def _llm_market_summary(self, context: FundamentalContext) -> str:
"""1-2 sentence macro blurb for alerts; "" when unavailable.
Cached (success OR failure) for the context-cache TTL so at most
one call is made per window."""
with NewsEngine._cache_lock:
ts, cached = NewsEngine._llm_summary_cache
if time.time() - ts < self.config.context_cache_ttl_seconds:
return cached
high_impact = [e for e in context.events_today if e.is_high_impact]
if not context.news_headlines and not high_impact:
return ""
client = get_llm_client()
if not client.available:
return ""
result = client.chat_json(_LLM_SUMMARY_SYSTEM, {
"headlines": [h.headline for h in context.news_headlines[:15]],
"events": [
{"event": e.event, "currency": e.currency,
"actual": e.actual, "forecast": e.forecast,
"released": not e.is_upcoming}
for e in high_impact[:10]
],
})
summary = ""
if result:
summary = str(result.get("summary", "")).strip()[:300]
with NewsEngine._cache_lock:
NewsEngine._llm_summary_cache = (time.time(), summary)
return summary
def _extract_currencies(self, headline: str) -> List[str]: def _extract_currencies(self, headline: str) -> List[str]:
"""Extract mentioned currencies from a headline.""" """Extract mentioned currencies from a headline."""
headline_lower = headline.lower() headline_lower = headline.lower()
+10 -7
View File
@@ -6,12 +6,13 @@ caching, validation, boundary guarantees, and range pagination — the
behavior the signal pipeline depends on). behavior the signal pipeline depends on).
Adding a provider = one module implementing `_fetch_raw` + one registry Adding a provider = one module implementing `_fetch_raw` + one registry
entry + one config.json `data_providers` block. The live feed is selected entry + one config.json `data_providers` block. The live candle feed is
by config.json `data_providers.active`; registration alone changes selected by config.json `data_providers.active`; the optional
nothing. `data_providers.streaming` scalar selects a different provider for the
tick stream (level-breach monitor). Registration alone changes nothing.
""" """
from typing import Dict from typing import Dict, Optional
from app_config import AppConfigError, get_app_config from app_config import AppConfigError, get_app_config
from config import PlatformConfig from config import PlatformConfig
@@ -29,9 +30,11 @@ PROVIDER_REGISTRY: Dict[str, type] = {
} }
def make_data_provider(config: PlatformConfig) -> BaseDataProvider: def make_data_provider(config: PlatformConfig,
"""Build the ACTIVE data provider (config.json: data_providers.active).""" name: Optional[str] = None) -> BaseDataProvider:
name = get_app_config().active_provider """Build a provider by name (default: config.json data_providers.active)."""
if name is None:
name = get_app_config().active_provider
cls = PROVIDER_REGISTRY.get(name) cls = PROVIDER_REGISTRY.get(name)
if cls is None: if cls is None:
raise AppConfigError( raise AppConfigError(
+109 -3
View File
@@ -10,15 +10,18 @@ Tiingo's JSON into the normalized frame.
Every failure mode raises `TiingoDataError` (a `DataIntegrityError`) Every failure mode raises `TiingoDataError` (a `DataIntegrityError`)
with a specific message — the scan loops log it and carry on; nothing with a specific message — the scan loops log it and carry on; nothing
fails silently. NOTE: this is the CANDLE feed only. Tiingo's free fails silently. NOTE: candles and mid-price ticks only. Tiingo's free
`/tiingo/fx/top` quote endpoint publishes indicative two-tier spreads `/tiingo/fx/top` quote endpoint publishes indicative two-tier spreads
and must not be used for spread display (removed 2026-07-23) — and must not be used for spread display (removed 2026-07-23) —
`get_spread` deliberately stays on the base-class default. `get_spread` deliberately stays on the base-class default, and the
WebSocket stream (`iter_ticks`) exposes the mid price only, never
bid/ask or a derived spread.
""" """
import json
import logging import logging
from datetime import datetime from datetime import datetime
from typing import Optional from typing import AsyncIterator, Callable, Optional, Sequence, Tuple
import pandas as pd import pandas as pd
import requests import requests
@@ -198,3 +201,106 @@ class TiingoDataProvider(BaseDataProvider):
"Fetched %d bars for %s @ %s from Tiingo", "Fetched %d bars for %s @ %s from Tiingo",
len(df), ticker, resample) len(df), ticker, resample)
return df return df
# ── Streaming quotes ───────────────────────────────────────────
supports_streaming = True
WS_URL = "wss://api.tiingo.com/fx"
@staticmethod
def _parse_ws_message(raw: str) -> Optional[Tuple[str, float]]:
"""Parse one WS message; return (tiingo_ticker, mid_price) for
quote updates, None for heartbeats ("H"), subscription acks
("I"), malformed JSON, or quotes without a usable price.
Server error messages ("E", e.g. bad token) RAISE TiingoDataError:
Tiingo authenticates inside the socket via the subscribe message,
so an auth failure arrives as a data message followed by a clean
close — swallowing it would make the caller's reconnect loop
hammer the endpoint with no backoff (backoff only grows on
exceptions).
"""
try:
msg = json.loads(raw)
except json.JSONDecodeError:
logger.warning("Tiingo WS: malformed JSON, skipping")
return None
mtype = msg.get("messageType")
if mtype == "H":
return None
if mtype == "I":
logger.info("Tiingo WS: subscription acknowledged")
return None
if mtype == "E":
raise TiingoDataError(
f"Tiingo WS server error: {str(msg.get('response'))[:200]}")
if mtype != "A":
return None
data = msg.get("data")
if not isinstance(data, list) or len(data) < 8 or data[0] != "Q":
return None
# Live-verified field order (2026-08-03): ["Q", ticker, timestamp,
# bidSize, bid, mid, askSize, ask] — ask is index 7, NOT 6 as the
# docs' table suggests (index 6 is askSize).
ticker = data[1]
bid, mid, ask = data[4], data[5], data[7]
if mid is None and bid is not None and ask is not None:
mid = (bid + ask) / 2
try:
return ticker, float(mid)
except (TypeError, ValueError):
logger.warning("Tiingo WS: quote for %s has no usable price",
ticker)
return None
async def iter_ticks(
self,
symbols: Sequence[str],
on_connect: Optional[Callable[[], None]] = None,
) -> AsyncIterator[Tuple[str, float]]:
"""One connection's lifetime of (canonical_symbol, mid) ticks.
The caller owns the reconnect/backoff loop around this generator.
Divergences from the Twelve Data reference implementation, both
deliberate: `on_connect` fires on the FIRST RECEIVED message (the
"I" ack in practice) rather than after the subscribe send, so a
rejected token cannot reset the caller's backoff; and there is no
client heartbeat task — Tiingo sends "H" frames itself and the
websockets library's built-in ping keepalive detects dead sockets.
"""
import websockets
if self._api_key is None:
raise TiingoDataError(
"No Tiingo API key. Set the TIINGO_API_KEY env var or "
"data_providers.tiingo.api_key in config.json.")
ws_to_pair = {self._to_tiingo_ticker(s): s for s in symbols}
async with websockets.connect(self.WS_URL) as ws:
await ws.send(json.dumps({
"eventName": "subscribe",
"authorization": self._api_key,
"eventData": {
"thresholdLevel": 5,
"tickers": sorted(ws_to_pair),
},
}))
logger.info(
"Tiingo WS connected, subscribing to %d tickers",
len(ws_to_pair))
connected = False
async for raw in ws:
parsed = self._parse_ws_message(raw)
if not connected:
connected = True
if on_connect is not None:
on_connect()
if parsed is None:
continue
pair_symbol = ws_to_pair.get(parsed[0])
if pair_symbol is None:
continue
yield pair_symbol, parsed[1]
+3
View File
@@ -79,6 +79,7 @@ class SignalReport:
event_description: str = "" event_description: str = ""
news_sentiment: float = 0.0 news_sentiment: float = 0.0
news_description: str = "" news_description: str = ""
news_summary: str = "" # LLM market-context blurb (display-only)
final_signal: SignalStrength = SignalStrength.NEUTRAL final_signal: SignalStrength = SignalStrength.NEUTRAL
final_score: float = 0.0 final_score: float = 0.0
@@ -832,11 +833,13 @@ class SignalAggregator:
sentiment, sent_desc = news.get_pair_sentiment(symbol, context) sentiment, sent_desc = news.get_pair_sentiment(symbol, context)
report.news_sentiment = sentiment report.news_sentiment = sentiment
report.news_description = sent_desc report.news_description = sent_desc
report.news_summary = context.llm_summary
else: else:
report.event_risk = "clear" report.event_risk = "clear"
report.event_description = "" report.event_description = ""
report.news_sentiment = 0.0 report.news_sentiment = 0.0
report.news_description = "" report.news_description = ""
report.news_summary = ""
# ── 5. Spread estimate from primary data (avoids extra API call) ─ # ── 5. Spread estimate from primary data (avoids extra API call) ─
try: try:
+36 -3
View File
@@ -12,7 +12,9 @@ Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set.
M15 fires at :00/:15/:30/:45 + 10s buffer. M15 fires at :00/:15/:30/:45 + 10s buffer.
H1 fires at :00 + 10s buffer. H1 fires at :00 + 10s buffer.
H4 fires at :00 every 4 hours + 10s buffer. H4 fires at :00 every 4 hours + 10s buffer.
WebSocket: Streams live ticks from TwelveData for XAU/USD. WebSocket: Streams live ticks from the streaming provider — config.json
`data_providers.streaming` when set (e.g. tiingo), else the
active provider if it streams; auto-disabled otherwise.
When price crosses a key structure level (swing S/R, BB, When price crosses a key structure level (swing S/R, BB,
EMA200) cached from the last boundary scan, triggers an EMA200) cached from the last boundary scan, triggers an
immediate M15 scan and sends a Level Break Alert. immediate M15 scan and sends a Level Break Alert.
@@ -203,6 +205,27 @@ def _get_shared_dp() -> BaseDataProvider:
return _shared_dp return _shared_dp
_streaming_dp: Optional[BaseDataProvider] = None
def _get_streaming_dp() -> BaseDataProvider:
"""Provider for the tick stream: config.json data_providers.streaming
when set (a SEPARATE instance — the candle cache stays on the active
provider), else the shared active provider. Reads app config at call
time, not the import-time _APP snapshot, so tests can swap it via
set_app_config()."""
global _streaming_dp
if _streaming_dp is None:
app = get_app_config()
if app.streaming_provider is None or \
app.streaming_provider == app.active_provider:
_streaming_dp = _get_shared_dp()
else:
_streaming_dp = make_data_provider(
PlatformConfig(), app.streaming_provider)
return _streaming_dp
# ── Signal formatting ───────────────────────────────────────────────── # ── Signal formatting ─────────────────────────────────────────────────
_SIGNAL_EMOJI = { _SIGNAL_EMOJI = {
@@ -372,6 +395,8 @@ def format_signal_message(report: SignalReport) -> str:
extra.append(f"📋 {ind_summary}") extra.append(f"📋 {ind_summary}")
for w in report.warnings: for w in report.warnings:
extra.append(f"⚠️ {_escape_md(w)}") extra.append(f"⚠️ {_escape_md(w)}")
if report.news_summary:
extra.append(f"📰 {_escape_md(report.news_summary)}")
if extra: if extra:
lines += [""] + extra lines += [""] + extra
@@ -755,12 +780,19 @@ async def _ws_monitor(bot) -> None:
"""Level-breach tick monitor. The provider owns the connection and """Level-breach tick monitor. The provider owns the connection and
protocol (`iter_ticks`); this loop owns reconnect/backoff and breach protocol (`iter_ticks`); this loop owns reconnect/backoff and breach
dispatch. Providers without streaming quotes simply disable it.""" dispatch. Providers without streaming quotes simply disable it."""
dp = _get_shared_dp() dp = _get_streaming_dp()
if not dp.supports_streaming: if not dp.supports_streaming:
logger.info( logger.info(
"Data provider %r has no streaming quotes — " "Data provider %r has no streaming quotes — "
"level-breach monitor disabled", dp.name) "level-breach monitor disabled", dp.name)
return return
if getattr(dp, "_api_key", "unused") is None:
# Guard, not a traceback loop: without a key every connect
# attempt would raise and re-log a full traceback every ≤60s.
logger.warning(
"Streaming provider %r has no API key — "
"level-breach monitor disabled", dp.name)
return
backoff = 1 backoff = 1
while _feed_active(): while _feed_active():
@@ -951,7 +983,8 @@ def _status_text() -> str:
lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)") lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)")
if _ws_task and not _ws_task.done(): if _ws_task and not _ws_task.done():
lines.append("📡 WebSocket: *connected*") provider = _streaming_dp.name if _streaming_dp else "?"
lines.append(f"📡 WebSocket: *connected* ({provider})")
else: else:
lines.append("❌ WebSocket: *disconnected*") lines.append("❌ WebSocket: *disconnected*")
+445 -2
View File
@@ -26,7 +26,7 @@ sys.path.insert(0, os.path.dirname(__file__))
import json import json
import app_config import app_config
for _k in ("TELEGRAM_TOKEN", "TELEGRAM_CHAT_ID", "TELEGRAM_CHANNEL_ID", for _k in ("TELEGRAM_TOKEN", "TELEGRAM_CHAT_ID", "TELEGRAM_CHANNEL_ID",
"TWELVE_DATA_API_KEY"): "TWELVE_DATA_API_KEY", "DEEPSEEK_API_KEY"):
os.environ.pop(_k, None) os.environ.pop(_k, None)
_TEST_APP = app_config.AppConfig( _TEST_APP = app_config.AppConfig(
provider_settings={"twelve_data": {"api_key": "test-key"}} provider_settings={"twelve_data": {"api_key": "test-key"}}
@@ -3001,6 +3001,85 @@ def _():
set_app_config(_TEST_APP) set_app_config(_TEST_APP)
@test("providers: make_data_provider explicit name overrides active")
def _():
from providers import (TiingoDataProvider as _Tii,
TwelveDataProvider, make_data_provider)
# active is twelve_data under _TEST_APP; explicit name wins
with patch.dict(os.environ, {"TIINGO_API_KEY": "k"}):
assert isinstance(make_data_provider(PlatformConfig(), "tiingo"), _Tii)
assert isinstance(make_data_provider(PlatformConfig()), TwelveDataProvider)
try:
make_data_provider(PlatformConfig(), "polygon")
assert False, "unregistered explicit name must raise"
except AppConfigError as e:
assert "polygon" in str(e)
@test("app_config: data_providers.streaming scalar selects the tick stream")
def _():
app = load_app_config(_write_json({
"data_providers": {"active": "oanda", "streaming": "tiingo",
"oanda": {}, "tiingo": {"api_key": "k"}}}))
assert app.streaming_provider == "tiingo"
assert set(app.provider_settings) == {"oanda", "tiingo"}, \
"reserved scalars must not become provider blocks"
app = load_app_config(_write_json(
{"data_providers": {"active": "oanda", "oanda": {}}}))
assert app.streaming_provider is None
@test("app_config: streaming provider must be a string with a settings block")
def _():
for payload in (
{"data_providers": {"active": "oanda", "oanda": {},
"streaming": "polygon"}}, # no block
{"data_providers": {"active": "oanda", "oanda": {},
"streaming": 5}}, # not a string
):
try:
load_app_config(_write_json(payload))
assert False, f"should have raised for {payload}"
except AppConfigError:
pass
@test("app_config: deepseek block parses, validates, defaults empty")
def _():
app = load_app_config(_write_json({"deepseek": {
"api_key": "sk-x", "model": "deepseek-v4-flash", "enabled": False,
"timeout_seconds": 5, "daily_request_cap": 100}}))
assert app.deepseek_settings["api_key"] == "sk-x"
assert app.deepseek_settings["enabled"] is False
assert load_app_config(_write_json({})).deepseek_settings == {}
for payload in (
{"deepseek": {"api_keyy": "typo"}}, # unknown key
{"deepseek": {"enabled": "yes"}}, # not a bool
{"deepseek": {"timeout_seconds": 0}}, # not positive
{"deepseek": {"timeout_seconds": True}}, # bool is not a number
{"deepseek": {"daily_request_cap": 2.5}}, # not an int
{"deepseek": {"model": ""}}, # empty model
{"deepseek": "oops"}, # not an object
):
try:
load_app_config(_write_json(payload))
assert False, f"should have raised for {payload}"
except AppConfigError:
pass
@test("app_config: deepseek_api_key resolves env → config.json → empty")
def _():
from app_config import deepseek_api_key
app = load_app_config(_write_json({"deepseek": {"api_key": "json-ds"}}))
with patch.dict(os.environ, {}, clear=True):
assert deepseek_api_key(app) == "json-ds"
with patch.dict(os.environ, {"DEEPSEEK_API_KEY": "env-ds"}):
assert deepseek_api_key(app) == "env-ds"
with patch.dict(os.environ, {}, clear=True):
assert deepseek_api_key(AppConfig()) == ""
@test("config.py: hardcoded twelve_data_api_key field is gone") @test("config.py: hardcoded twelve_data_api_key field is gone")
def _(): def _():
assert not hasattr(PlatformConfig(), "twelve_data_api_key"), \ assert not hasattr(PlatformConfig(), "twelve_data_api_key"), \
@@ -3126,6 +3205,55 @@ def _():
tb._autoscan_chats.clear() tb._autoscan_chats.clear()
@test("telegram_bot: _get_streaming_dp resolves streaming name, else shared")
def _():
import telegram_bot as tb
from providers import TiingoDataProvider as _Tii
set_app_config(AppConfig(
active_provider="twelve_data",
streaming_provider="tiingo",
provider_settings={"twelve_data": {"api_key": "k"},
"tiingo": {"api_key": "k"}}))
tb._streaming_dp = None
tb._shared_dp = None
try:
dp = tb._get_streaming_dp()
assert isinstance(dp, _Tii)
assert dp is not tb._get_shared_dp(), \
"streaming provider must be a separate instance"
# unset → falls back to the shared active provider
set_app_config(_TEST_APP)
tb._streaming_dp = None
tb._shared_dp = None
assert tb._get_streaming_dp() is tb._get_shared_dp()
finally:
set_app_config(_TEST_APP)
tb._streaming_dp = None
tb._shared_dp = None
@test("telegram_bot: ws monitor auto-disables on non-streaming or keyless")
def _():
import asyncio
import telegram_bot as tb
class NoStream:
name = "stub"
supports_streaming = False
class KeylessStream:
name = "stub"
supports_streaming = True
_api_key = None
try:
for stub in (NoStream(), KeylessStream()):
tb._streaming_dp = stub
asyncio.run(tb._ws_monitor(None)) # must return, not connect
finally:
tb._streaming_dp = None
# ══════════════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════════════
# MODULE 15: Oanda provider / executor / daemon # MODULE 15: Oanda provider / executor / daemon
# ══════════════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════════════
@@ -4062,7 +4190,7 @@ def _():
def _(): def _():
assert TwelveDataProvider.supports_streaming is True assert TwelveDataProvider.supports_streaming is True
assert _POanda.supports_streaming is False assert _POanda.supports_streaming is False
assert TiingoDataProvider.supports_streaming is False assert TiingoDataProvider.supports_streaming is True
class Minimal(BaseDataProvider): class Minimal(BaseDataProvider):
name = "minimal" name = "minimal"
@@ -4092,6 +4220,49 @@ def _():
assert parse('{"event":"price","symbol":"XAU/USD","price":"abc"}') is None assert parse('{"event":"price","symbol":"XAU/USD","price":"abc"}') is None
@test("providers: Tiingo WS quote-message parsing")
def _():
parse = TiingoDataProvider._parse_ws_message
# Field order matches a live-captured frame (2026-08-03):
# ["Q", ticker, ts, bidSize, bid, mid, askSize, ask]
quote = ('{"service":"fx","messageType":"A","data":'
'["Q","xauusd","2026-08-03T17:09:46.084000+00:00",'
'1000000.0,4035.05,4035.1,1000000.0,4035.15]}')
assert parse(quote) == ("xauusd", 4035.1)
# mid null → (bid+ask)/2, ask read from index 7 (index 6 is askSize)
mid_null = ('{"messageType":"A","service":"fx","data":'
'["Q","xauusd","2026-08-03T10:00:00+00:00",'
'1000000,3344.0,null,1000000,3346.0]}')
assert parse(mid_null) == ("xauusd", 3345.0)
# mid and bid null → unusable, skipped
all_null = ('{"messageType":"A","service":"fx","data":'
'["Q","xauusd","2026-08-03T10:00:00+00:00",'
'1000000,null,null,1000000,3346.0]}')
assert parse(all_null) is None
assert parse('{"messageType":"H"}') is None # heartbeat
assert parse('{"messageType":"I","response":{"code":200}}') is None
assert parse('not json{{') is None
assert parse('{"messageType":"A","data":["X","xauusd"]}') is None
# server error (e.g. bad token) must RAISE so the caller's backoff
# applies — Tiingo auths inside the socket, not at the HTTP upgrade
try:
parse('{"messageType":"E","response":'
'{"code":401,"message":"bad token"}}')
assert False, "server error message must raise"
except TiingoDataError:
pass
@test("providers: Tiingo WS ticker mapping round-trips the scanned pairs")
def _():
tp = TiingoDataProvider.__new__(TiingoDataProvider)
for pair in PAIR_INFO:
ticker = tp._to_tiingo_ticker(pair)
assert ticker.islower() and "/" not in ticker
inverse = {tp._to_tiingo_ticker(s): s for s in PAIR_INFO}
assert inverse[ticker] == pair
@test("providers: Oanda date-window fetch omits count and sends from/to") @test("providers: Oanda date-window fetch omits count and sends from/to")
def _(): def _():
captured = {} captured = {}
@@ -4522,6 +4693,278 @@ def _():
assert hits == ["M5_SELL_QUALITY", "SELL_HOUR", "TREND_EXHAUSTION"], hits assert hits == ["M5_SELL_QUALITY", "SELL_HOUR", "TREND_EXHAUSTION"], hits
# ══════════════════════════════════════════════════════════════════════
# MODULE 19: DeepSeek news enrichment (llm_client + news_engine seams)
# ══════════════════════════════════════════════════════════════════════
print("\n═══ Module 19: DeepSeek news enrichment ═══")
import datetime as _dt_m19
import llm_client as lc
from llm_client import DeepSeekClient
from news_engine import (EconomicEvent, FundamentalContext, NewsEngine,
NewsSentiment)
def _stub_resp(payload, status=200):
return type("R", (), {
"status_code": status,
"text": json.dumps(payload),
"json": lambda self: payload,
})()
def _chat_body(content_obj):
return {"choices": [{"message": {"content": json.dumps(content_obj)}}]}
def _counting_session(payload, status=200):
class S:
calls = 0
def post(self, url, json=None, timeout=None):
S.calls += 1
return _stub_resp(payload, status)
return S()
@test("llm_client: response parsing and availability gating")
def _():
parse = DeepSeekClient._parse_response
assert parse(_chat_body({"a": 1})) == {"a": 1}
assert parse({"choices": []}) is None
assert parse({"error": "x"}) is None
assert parse({"choices": [{"message": {"content": "not json"}}]}) is None
assert parse({"choices": [{"message": {"content": "[1,2]"}}]}) is None
# keyless / disabled clients never touch the network
for client in (DeepSeekClient(api_key=""),
DeepSeekClient(api_key="k", enabled=False)):
assert client.available is False
s = _counting_session(_chat_body({"a": 1}))
client._session = s
assert client.chat_json("sys", {"q": 1}) is None
assert type(s).calls == 0
@test("llm_client: daily cap blocks requests and resets on UTC date roll")
def _():
client = DeepSeekClient(api_key="k", daily_cap=2)
s = _counting_session(_chat_body({"ok": True}))
client._session = s
assert client.chat_json("sys", {}) == {"ok": True}
assert client.chat_json("sys", {}) == {"ok": True}
assert client.chat_json("sys", {}) is None, "third call must hit the cap"
assert type(s).calls == 2, "capped call must make no HTTP request"
# UTC date roll resets the counter
client._count_day = (_dt_m19.datetime.now(_dt_m19.timezone.utc).date()
- _dt_m19.timedelta(days=1))
assert client.chat_json("sys", {}) == {"ok": True}
assert type(s).calls == 3
@test("llm_client: HTTP error and malformed body return None")
def _():
client = DeepSeekClient(api_key="k")
client._session = _counting_session({"detail": "quota"}, status=429)
assert client.chat_json("sys", {}) is None
client._session = _counting_session({"unexpected": "shape"})
assert client.chat_json("sys", {}) is None
class _StubLLM:
"""Drop-in for the llm_client singleton."""
def __init__(self, reply):
self.reply = reply
self.calls = 0
available = True
def chat_json(self, system_prompt, user_payload):
self.calls += 1
return self.reply
@test("NewsEngine: _llm_headline_scores validates, clamps, rejects mismatch")
def _():
ne = NewsEngine(NewsConfig())
try:
lc._client = _StubLLM({"scores": [[5, -3], [0.2, 0.1]]})
assert ne._llm_headline_scores(["a", "b"]) == [(1.0, -1.0), (0.2, 0.1)]
lc._client = _StubLLM({"scores": [[0.1, 0.1]]})
assert ne._llm_headline_scores(["a", "b"]) is None, "length mismatch"
lc._client = _StubLLM({"scores": [["hot", 0], [0.3]]})
assert ne._llm_headline_scores(["a", "b"]) is None, "non-numeric/short"
lc._client = _StubLLM(None)
assert ne._llm_headline_scores(["a"]) is None, "failed call"
finally:
lc._client = None
@test("NewsEngine: LLM sentiment overrides keyword scores per currency")
def _():
now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
ne = NewsEngine(NewsConfig())
NewsEngine._llm_sent_cache = {}
try:
lc._client = _StubLLM({"ok": True}) # available, unused (stubbed below)
h = NewsSentiment(
headline="Fed signals rate hike", source="s", timestamp=now,
currencies_mentioned=["USD", "XAU"], sentiment_score=1.0,
keywords=["rate hike"],
sentiment_by_currency={"USD": 1.0, "XAU": -1.0})
ne._llm_headline_scores = lambda texts: [(0.4, -0.7)] * len(texts)
ne._apply_llm_sentiment([h])
assert h.sentiment_score == 0.4
assert h.sentiment_by_currency == {"USD": 0.4, "XAU": -0.7}
assert h.keywords == ["rate hike"], "keywords stay keyword-derived"
finally:
NewsEngine._llm_sent_cache = {}
lc._client = None
@test("NewsEngine: LLM failure keeps keyword sentiment untouched")
def _():
now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
ne = NewsEngine(NewsConfig())
NewsEngine._llm_sent_cache = {}
try:
lc._client = _StubLLM({"ok": True})
h = NewsSentiment(
headline="Fed signals rate hike", source="s", timestamp=now,
currencies_mentioned=["USD"], sentiment_score=0.6,
keywords=[], sentiment_by_currency={"USD": 0.6})
ne._llm_headline_scores = lambda texts: None
ne._apply_llm_sentiment([h])
assert h.sentiment_score == 0.6
assert h.sentiment_by_currency == {"USD": 0.6}
# unavailable client: not even the scores call happens
lc._client = None
ne._llm_headline_scores = lambda texts: (_ for _ in ()).throw(
AssertionError("must not be called when unavailable"))
ne._apply_llm_sentiment([h])
finally:
NewsEngine._llm_sent_cache = {}
lc._client = None
@test("NewsEngine: LLM sentiment cache only sends new headlines")
def _():
now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
ne = NewsEngine(NewsConfig())
NewsEngine._llm_sent_cache = {}
sent_batches = []
def make(h):
return NewsSentiment(
headline=h, source="s", timestamp=now,
currencies_mentioned=["XAU"], sentiment_score=0.0,
keywords=[], sentiment_by_currency={"XAU": 0.0})
try:
lc._client = _StubLLM({"ok": True})
def scores(texts):
sent_batches.append(list(texts))
return [(0.5, 0.5)] * len(texts)
ne._llm_headline_scores = scores
ne._apply_llm_sentiment([make("headline one"), make("headline two")])
ne._apply_llm_sentiment([make("headline one"), make("headline two"),
make("headline three")])
assert sent_batches == [["headline one", "headline two"],
["headline three"]], sent_batches
finally:
NewsEngine._llm_sent_cache = {}
lc._client = None
@test("NewsEngine: LLM enriches surprise description, numeric fallback intact")
def _():
now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
ne = NewsEngine(NewsConfig())
NewsEngine._llm_event_cache = {}
released = EconomicEvent(
time=now - _dt_m19.timedelta(minutes=10), currency="USD",
impact="high", event="Nonfarm Payrolls",
forecast="200K", actual="300K")
ctx = FundamentalContext(events_today=[released])
try:
# failure first: description must equal today's numeric form
lc._client = _StubLLM(None)
risk, desc = ne.get_event_risk_level("XAU/USD", ctx)
assert risk == "caution"
assert desc.endswith("(actual 300K vs forecast 200K)"), desc
# then success: LLM direction appended to the description only
lc._client = _StubLLM({"gold_direction": "bearish",
"reason": "strong jobs lift yields"})
risk, desc = ne.get_event_risk_level("XAU/USD", ctx)
assert risk == "caution"
assert desc.endswith("— gold bearish: strong jobs lift yields"), desc
# bogus direction → numeric fallback
NewsEngine._llm_event_cache = {}
lc._client = _StubLLM({"gold_direction": "sideways"})
_, desc = ne.get_event_risk_level("XAU/USD", ctx)
assert desc.endswith("(actual 300K vs forecast 200K)"), desc
finally:
NewsEngine._llm_event_cache = {}
lc._client = None
@test("NewsEngine: market summary cached across engine instances")
def _():
now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
NewsEngine._llm_summary_cache = (0.0, "")
ctx = FundamentalContext(news_headlines=[NewsSentiment(
headline="Gold steady", source="s", timestamp=now,
currencies_mentioned=["XAU"], sentiment_score=0.0, keywords=[])])
try:
stub = _StubLLM({"summary": "Soft dollar supports gold."})
lc._client = stub
assert NewsEngine(NewsConfig())._llm_market_summary(ctx) == \
"Soft dollar supports gold."
# second engine instance within the TTL: served from cache, no call
assert NewsEngine(NewsConfig())._llm_market_summary(ctx) == \
"Soft dollar supports gold."
assert stub.calls == 1
# failure caches "" so a broken API is retried next window, not per scan
NewsEngine._llm_summary_cache = (0.0, "")
lc._client = _StubLLM(None)
assert NewsEngine(NewsConfig())._llm_market_summary(ctx) == ""
assert NewsEngine._llm_summary_cache[1] == ""
finally:
NewsEngine._llm_summary_cache = (0.0, "")
lc._client = None
@test("SignalReport.news_summary flows to Telegram, resets when news off")
def _():
import telegram_bot as tb
config = PlatformConfig()
config.primary_timeframe = Timeframe.H1
config.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
agg = SignalAggregator(config)
dfs = {
Timeframe.H1: make_ohlcv(300, seed=42),
Timeframe.M15: make_ohlcv(300, seed=43),
Timeframe.H4: make_ohlcv(300, seed=44),
}
ctx = FundamentalContext(llm_summary="Test macro blurb.")
report = agg.analyze_from_dataframes(
"XAU/USD", dfs, context=ctx, use_news=True)
assert report.news_summary == "Test macro blurb."
assert "Test macro blurb." in tb.format_signal_message(report)
report = agg.analyze_from_dataframes(
"XAU/USD", dfs, context=ctx, use_news=False)
assert report.news_summary == ""
@test("news defaults: LLM never gates signals (2026-07 audit lock)")
def _():
cfg = NewsConfig()
assert cfg.event_risk_caps_signals is False
assert cfg.sentiment_affects_signals is False
# ══════════════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════════════
# SUMMARY # SUMMARY
# ══════════════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════════════