From d669227ddc673df84584d44a3548c967d2c07899 Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Fri, 24 Jul 2026 14:10:45 -0400 Subject: [PATCH] Oanda demo daemon: auto-execution on practice account (XAU/USD) New independent trading daemon (oanda_daemon.py) that places orders on an Oanda v20 practice account instead of sending Telegram alerts. Oanda-only market data (oanda_provider.py, mid candles + real bid/ask spread) through the unchanged signal pipeline; order execution and trade management via oanda_executor.py (raw requests, no new deps). Book parity with the backtester: no broker-side TP (exits = SL/TRAIL/ TIMEOUT only), bar-close-only trail ratchet with each closed bar processed exactly once, 5-bar per-TF entry cooldown, structural SL from the setup. Accepted divergence: one position total (skipped signals logged for book reconciliation). Safety/ops: fail-fast boot on rejected credentials (transient outages continue), market orders never retried (double-fill risk), entry lock against aligned-boundary races, restart recovery + adoption via oanda_state.json, hourly heartbeat, asyncio exception handler, every failure path logged. Prices formatted at instrument precision. Config: env OANDA_* -> config.json "oanda" section (validated in app_config). Tests: Module 15 (30 tests), suite 240 green. Claude-Session: https://claude.ai/code/session_01VUnCqjWBuvKRSjLwHeB9BJ --- .gitignore | 3 + app_config.py | 45 ++- config.json.example | 8 + oanda_daemon.py | 775 ++++++++++++++++++++++++++++++++++++++++ oanda_executor.py | 212 +++++++++++ oanda_provider.py | 361 +++++++++++++++++++ test_all.py | 842 ++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 2245 insertions(+), 1 deletion(-) create mode 100644 oanda_daemon.py create mode 100644 oanda_executor.py create mode 100644 oanda_provider.py diff --git a/.gitignore b/.gitignore index 22998c1..69b0ff9 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,9 @@ .env.local config.json +# Oanda daemon state +oanda_state.json + # Editor / tooling .claude/ .vscode/ diff --git a/app_config.py b/app_config.py index 7aa94d0..e4cb012 100644 --- a/app_config.py +++ b/app_config.py @@ -85,7 +85,11 @@ _VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") # AUTOSCAN_CONFIRMATIONS entries in telegram_bot.py). _SCANNABLE_TFS = ("M5", "M15", "H1", "H4") -_KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level"} +_KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level", "oanda"} +_KNOWN_OANDA = { + "account_id", "api_token", "base_url", "lot_size", + "quality_gate", "max_hold_bars", +} _KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"} _KNOWN_SCANNING = {"pairs", "timeframes"} _KNOWN_PAIR_KEYS = {"blocked_hours_utc"} @@ -107,6 +111,7 @@ class AppConfig: scan_timeframes: List[Timeframe] = field(default_factory=lambda: [ Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]) pair_overrides: Dict[str, Dict[str, Any]] = field(default_factory=dict) + oanda_settings: Dict[str, Any] = field(default_factory=dict) log_level: str = "INFO" @@ -212,6 +217,24 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig: "be a list of integers in 0-23") app.pair_overrides = {s: dict(o) for s, o in pair_section.items()} + oanda = raw.get("oanda", {}) + _check_keys(oanda, _KNOWN_OANDA, '"oanda"') + if "lot_size" in oanda: + ls = oanda["lot_size"] + if not isinstance(ls, int) or isinstance(ls, bool) or ls < 1: + raise AppConfigError( + "config.json: oanda.lot_size must be a positive integer") + if "quality_gate" in oanda and oanda["quality_gate"] not in _VALID_QUALITIES: + raise AppConfigError( + f'config.json: oanda.quality_gate "{oanda["quality_gate"]}" ' + f"not in {_VALID_QUALITIES}") + if "max_hold_bars" in oanda: + mhb = oanda["max_hold_bars"] + if not isinstance(mhb, int) or isinstance(mhb, bool) or mhb < 1: + raise AppConfigError( + "config.json: oanda.max_hold_bars must be a positive integer") + app.oanda_settings = dict(oanda) + level = raw.get("log_level", app.log_level) if level not in _VALID_LOG_LEVELS: raise AppConfigError( @@ -271,6 +294,26 @@ def provider_api_key(app: Optional[AppConfig] = None) -> str: return str(block.get("api_key", "")) +# ── Oanda resolution helpers (env → config.json → default) ─────────── + +def oanda_account_id(app: Optional[AppConfig] = None) -> str: + app = app or get_app_config() + return os.environ.get("OANDA_ACCOUNT_ID") or str( + app.oanda_settings.get("account_id", "")) + + +def oanda_api_token(app: Optional[AppConfig] = None) -> str: + app = app or get_app_config() + return os.environ.get("OANDA_API_TOKEN") or str( + app.oanda_settings.get("api_token", "")) + + +def oanda_base_url(app: Optional[AppConfig] = None) -> str: + app = app or get_app_config() + return os.environ.get("OANDA_BASE_URL") or str( + app.oanda_settings.get("base_url", "https://api-fxpractice.oanda.com")) + + # ── PlatformConfig overlay ──────────────────────────────────────────── def apply_pair_overrides( diff --git a/config.json.example b/config.json.example index 11fdbf4..261d631 100644 --- a/config.json.example +++ b/config.json.example @@ -20,5 +20,13 @@ "blocked_hours_utc": [0, 1, 2, 10, 11, 13, 16, 21] } }, + "oanda": { + "account_id": "REPLACE_WITH_ACCOUNT_ID", + "api_token": "REPLACE_WITH_API_TOKEN", + "base_url": "https://api-fxpractice.oanda.com", + "lot_size": 1, + "quality_gate": "C", + "max_hold_bars": 300 + }, "log_level": "INFO" } diff --git a/oanda_daemon.py b/oanda_daemon.py new file mode 100644 index 0000000..8dd19a9 --- /dev/null +++ b/oanda_daemon.py @@ -0,0 +1,775 @@ +""" +ForexBot2 — Oanda Automated Trading Daemon +============================================ +Autonomous trading process for XAU/USD on an Oanda practice account. +Runs candle-boundary scan loops (M5/M15/H1/H4), executes trades via +the Oanda v20 API, and manages trailing stops at bar close — matching +the backtested bar-close ratchet model. + +Book-parity notes (matching backtest/gpu_backtest.py): +- No broker-side take-profit: in the trailing model TP1/TP2 are + advisory; the only exits are SL (broker), TRAIL (daemon ratchet), + and TIMEOUT (max_hold_bars). A hard TP would cap winners the trail + lets run and diverge the live book from the backtest. +- Entry cooldown is 5 bars per timeframe (the backtester's only + sequential rule) — not a fingerprint block. +- The trail ratchets at bar CLOSE only, and each closed bar is + processed exactly once (last_bar_ts guard), so weekends and feed + lag can't double-count bars or inflate the hold counter. + +Divergence from the backtest book (accepted for demo safety): one +position total. While a trade is open, other signals are logged and +skipped, so skipped entries can be compared against the book later. + +Usage: python oanda_daemon.py +Config: env vars OANDA_ACCOUNT_ID, OANDA_API_TOKEN, OANDA_BASE_URL + or config.json "oanda" section (env wins). +""" + +import asyncio +import json +import logging +import os +import signal +import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from logging.handlers import TimedRotatingFileHandler +from typing import Dict, Optional + +import pandas as pd + +from app_config import ( + get_app_config, oanda_account_id, oanda_api_token, oanda_base_url, +) +from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD +from signal_aggregator import SignalAggregator, SignalReport, SignalStrength +from pair_signals import PAIR_PRODUCERS +from oanda_provider import OandaDataProvider +from oanda_executor import OandaExecutor, OandaAPIError + +logger = logging.getLogger(__name__) + +# ── Constants ────────────────────────────────────────────────────────── + +SYMBOL = XAUUSD +INSTRUMENT = "XAU_USD" + +SCAN_CONFIRMATIONS = { + Timeframe.M5: [Timeframe.M15, Timeframe.H1], + Timeframe.M15: [Timeframe.M5, Timeframe.H1], + Timeframe.H1: [Timeframe.M15, Timeframe.H4], + Timeframe.H4: [Timeframe.H1, Timeframe.D1], +} + +TF_INTERVAL_MINUTES = { + Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30, + Timeframe.H1: 60, Timeframe.H4: 240, Timeframe.D1: 1440, +} + +SCAN_TIMEFRAMES = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4] + +QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1, "D": 0, "F": -1} + +# Matches the backtester's --cooldown-bars default (the only sequential +# rule in the replay): no new entry within 5 bars of the previous entry +# on the same timeframe. +COOLDOWN_BARS = 5 + +# How often the heartbeat confirms the process is alive and the account +# is still reachable — independent of the candle-boundary loops, so a +# wedged scan or mid-run credential expiry still surfaces in the log. +HEARTBEAT_SECONDS = 3600 + +STATE_FILE = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "oanda_state.json") + +# ── Config resolution (env → config.json → default) ────────────────── + +def _oanda_setting_int(env_key: str, json_key: str, default: int) -> int: + env = os.environ.get(env_key) + if env: + return int(env) + return int(get_app_config().oanda_settings.get(json_key, default)) + + +def _oanda_setting_str(env_key: str, json_key: str, default: str) -> str: + return (os.environ.get(env_key) + or str(get_app_config().oanda_settings.get(json_key, default))) + +# ── Logging setup ────────────────────────────────────────────────────── + +def _setup_logging(level: str = "INFO") -> None: + log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs") + os.makedirs(log_dir, exist_ok=True) + log_file = os.path.join(log_dir, "daemon.log") + + fmt = logging.Formatter( + "%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S") + + file_handler = TimedRotatingFileHandler( + log_file, when="h", interval=3, backupCount=240, utc=True) + file_handler.setFormatter(fmt) + + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setFormatter(fmt) + + logging.basicConfig( + level=getattr(logging, level.upper(), logging.INFO), + handlers=[file_handler, console_handler], + force=True) + +# ── Boundary timing (same as telegram_bot.py) ───────────────────────── + +def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float: + now = datetime.now(timezone.utc) + total_minutes = now.hour * 60 + now.minute + elapsed_minutes = total_minutes % interval_minutes + 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 + return remaining + +# ── Trade state persistence ─────────────────────────────────────────── + +def _load_state() -> Dict: + if os.path.exists(STATE_FILE): + try: + with open(STATE_FILE, encoding="utf-8") as fh: + return json.load(fh) + except (OSError, json.JSONDecodeError): + logger.warning("Could not load state file, starting fresh") + return {"trades": {}, "last_entries": {}} + + +def _save_state(state: Dict) -> None: + try: + tmp = STATE_FILE + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(state, fh, indent=2) + os.replace(tmp, STATE_FILE) + except OSError: + logger.exception("Failed to save state file") + +# ── Trading Daemon ───────────────────────────────────────────────────── + +class TradingDaemon: + + def __init__(self): + account_id = oanda_account_id() + api_token = oanda_api_token() + 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") + self._max_hold_bars = _oanda_setting_int( + "OANDA_MAX_HOLD_BARS", "max_hold_bars", 300) + self._price_precision = PAIR_INFO[SYMBOL]["price_precision"] + + if not account_id or not api_token: + msg = ( + "Missing Oanda credentials. Set OANDA_ACCOUNT_ID and " + "OANDA_API_TOKEN env vars or config.json oanda section.") + logger.critical(msg) + raise SystemExit(msg) + + self._provider = OandaDataProvider( + PlatformConfig(), api_token, base_url) + self._executor = OandaExecutor( + account_id, api_token, base_url, lot_size, + price_precision=self._price_precision) + self._thread_pool = ThreadPoolExecutor(max_workers=4) + + self._state = _load_state() + self._state.setdefault("trades", {}) + self._state.setdefault("last_entries", {}) + self._state_lock = threading.Lock() + # Serializes the position-check + order-placement critical section + # across concurrent TF loops (all four fire at aligned boundaries). + self._entry_lock = threading.Lock() + self._running = True + + # ── Main entry point ─────────────────────────────────────────── + + async def run(self) -> None: + loop = asyncio.get_running_loop() + loop.set_exception_handler(self._asyncio_exception_handler) + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, self._request_shutdown) + + logger.info("=== Oanda Trading Daemon starting ===") + logger.info( + "Quality gate: %s | Max hold: %d bars | Cooldown: %d bars | " + "Timeframes: %s", + self._quality_gate, self._max_hold_bars, COOLDOWN_BARS, + "/".join(tf.name for tf in SCAN_TIMEFRAMES)) + + # Credentials first: with a bad token every later call fails, so + # refuse to start before touching positions or spawning loops. + self._verify_account_access() + + self._discover_existing_positions() + + tasks = [] + for tf in SCAN_TIMEFRAMES: + interval = TF_INTERVAL_MINUTES[tf] + tasks.append(asyncio.create_task( + self._candle_loop(tf, interval))) + tasks.append(asyncio.create_task(self._heartbeat_loop())) + + logger.info("Candle-boundary loops started") + try: + await asyncio.gather(*tasks) + except asyncio.CancelledError: + pass + finally: + self._shutdown() + + def _verify_account_access(self) -> None: + """Boot-time credential check. A 4xx here means the token/account + id is wrong — that never fixes itself, so refuse to start rather + than run a daemon that can't place or manage trades. A connection + failure or 5xx is a transient outage: log it and continue, since + broker-side stops already protect any open position and the + hourly heartbeat will keep reporting account health.""" + try: + self._executor.get_account_summary() + except OandaAPIError as e: + if 400 <= e.status_code < 500: + logger.critical( + "Oanda rejected the account credentials (HTTP %d): " + "%s — refusing to start", e.status_code, e) + raise SystemExit( + f"Oanda rejected the account credentials " + f"(HTTP {e.status_code}): {e}") + logger.error( + "Failed to fetch account summary at boot (HTTP %d): %s", + e.status_code, e) + + def _asyncio_exception_handler(self, loop, context) -> None: + exc = context.get("exception") + if exc is not None: + logger.critical( + "Uncaught asyncio exception: %s", + context.get("message", ""), exc_info=exc) + else: + logger.critical( + "Uncaught asyncio error: %s", context.get("message", "")) + + def _request_shutdown(self) -> None: + logger.info("Shutdown requested") + self._running = False + for task in asyncio.all_tasks(): + task.cancel() + + def _shutdown(self) -> None: + with self._state_lock: + _save_state(self._state) + self._thread_pool.shutdown(wait=False) + logger.info("=== Oanda Trading Daemon stopped ===") + + # ── Position discovery (restart recovery) ────────────────────── + + def _discover_existing_positions(self) -> None: + try: + open_trades = self._executor.get_open_trades(INSTRUMENT) + except OandaAPIError: + logger.exception("Failed to query open trades at boot") + return + + saved = self._state.get("trades", {}) + live_ids = {t["id"] for t in open_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) + del saved[tid] + + for trade in open_trades: + tid = trade["id"] + if tid in saved: + logger.info( + "Recovered trade %s from state: %s %s @ %.2f, SL %.2f", + tid, saved[tid].get("direction"), INSTRUMENT, + saved[tid].get("entry"), saved[tid].get("current_sl")) + continue + + 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 + + saved[tid] = { + "tf": "H1", + "direction": direction, + "entry": entry, + "sl_initial": sl, + "current_sl": sl, + "risk": risk, + "peak": entry, + "trail_armed": False, + "trail_arm_r": 0.6, + "trail_dist_r": 0.75, + "bars_held": 0, + "last_bar_ts": None, + "signal_bar_ts": None, + } + logger.info( + "Adopted unknown trade %s: %s @ %.2f, SL %.2f, risk %.2f " + "(conservative defaults)", + tid, direction, entry, sl, risk) + + self._state["trades"] = saved + _save_state(self._state) + logger.info( + "Position discovery: %d active trade(s)", len(saved)) + + # ── Candle-boundary loop ─────────────────────────────────────── + + async def _candle_loop(self, tf: Timeframe, interval_minutes: int) -> None: + label = tf.value + try: + while self._running: + wait = _seconds_until_boundary(interval_minutes, buffer=10) + logger.info("%s loop: next scan in %.0fs", label, wait) + await asyncio.sleep(wait) + if not self._running: + break + logger.info("%s loop: firing scan", label) + try: + await self._scan_and_manage(tf) + except Exception: + logger.exception("%s loop: cycle failed", label) + except asyncio.CancelledError: + pass + finally: + logger.info("%s loop stopped", label) + + async def _scan_and_manage(self, tf: Timeframe) -> None: + loop = asyncio.get_running_loop() + + # Trade management FIRST: the SL ratchet must not wait on (or be + # starved by) the signal scan — news fetches can take seconds and + # a failed scan must never cost an open trade its trail update. + try: + await loop.run_in_executor( + self._thread_pool, self._manage_open_trades, tf) + except Exception: + logger.exception("%s: trade management failed", tf.value) + + report = await loop.run_in_executor( + self._thread_pool, self._run_signal, tf) + + if not self._should_execute(report, tf): + return + + await loop.run_in_executor( + self._thread_pool, self._execute_trade, report, tf) + + def _manage_open_trades(self, tf: Timeframe) -> None: + self._check_closed_trades() + self._manage_trailing(tf) + + # ── Heartbeat ───────────────────────────────────────────────── + + def _heartbeat_once(self) -> None: + with self._state_lock: + n = len(self._state.get("trades", {})) + try: + self._executor.get_account_summary() + except OandaAPIError as e: + logger.error("HEARTBEAT: account summary failed: %s", e) + # Logged unconditionally — even on a failed summary call the + # heartbeat line itself must reach the log, so a wedged process + # doesn't look identical to a healthy but silent one. + logger.info("HEARTBEAT: alive | tracked trades: %d", n) + + async def _heartbeat_loop(self) -> None: + loop = asyncio.get_running_loop() + try: + while self._running: + await asyncio.sleep(HEARTBEAT_SECONDS) + if not self._running: + break + try: + await loop.run_in_executor( + self._thread_pool, self._heartbeat_once) + except Exception: + logger.exception("heartbeat loop: cycle failed") + except asyncio.CancelledError: + pass + finally: + logger.info("heartbeat loop stopped") + + # ── Signal pipeline ──────────────────────────────────────────── + + def _run_signal(self, tf: Timeframe) -> SignalReport: + conf_tfs = SCAN_CONFIRMATIONS.get( + tf, [Timeframe.M15, Timeframe.H4]) + factory = PAIR_PRODUCERS.get(SYMBOL) + if factory is not None: + config = factory(tf, conf_tfs) + else: + config = PlatformConfig() + config.primary_timeframe = tf + config.confirmation_timeframes = conf_tfs + + aggregator = SignalAggregator(config) + aggregator.data_provider = self._provider + report = aggregator.analyze_pair(symbol=SYMBOL) + + logger.info( + "%s scan: signal=%s score=%.1f conf=%.0f quality=%s " + "regime=%s vol=%s price=%.2f spread=%.4f", + tf.value, + report.final_signal.name, report.final_score, + report.final_confidence, report.signal_quality, + report.market_regime, report.volatility_regime, + report.current_price, + self._provider.get_spread(SYMBOL)) + + # Stale-data/suppression warnings must be visible at WARNING; + # routine session-filter notices go to DEBUG — the M5 loop fires + # 288x/day and would drown the log at WARNING otherwise. + for w in report.warnings: + if "STALE" in w.upper() or "FAIL" in w.upper(): + logger.warning("%s scan: %s", tf.value, w) + else: + logger.debug("%s scan: %s", tf.value, w) + + return report + + # ── Execution decision ───────────────────────────────────────── + + def _should_execute(self, report: SignalReport, tf: Timeframe) -> bool: + if report.final_signal == SignalStrength.NEUTRAL: + return False + + if report.trade_setup is None: + return False + + quality_rank = QUALITY_RANK.get(report.signal_quality, -1) + gate_rank = QUALITY_RANK.get(self._quality_gate, 1) + if quality_rank < gate_rank: + return False + + if self._in_cooldown(tf): + logger.info( + "%s: actionable signal (%s %s) but inside %d-bar entry " + "cooldown, skipping", + tf.value, report.trade_setup.direction, + report.signal_quality, COOLDOWN_BARS) + return False + + with self._state_lock: + if self._state.get("trades"): + logger.info( + "%s: actionable signal (%s %s) but position already " + "open, skipping", + tf.value, report.trade_setup.direction, + report.signal_quality) + return False + + return True + + def _in_cooldown(self, tf: Timeframe) -> bool: + """Backtest parity: no entry within COOLDOWN_BARS bars of the + previous entry on the same timeframe (wall-clock approximation).""" + with self._state_lock: + last = self._state.get("last_entries", {}).get(tf.value) + if not last: + return False + try: + last_ts = datetime.fromisoformat(last) + except ValueError: + return False + interval_s = TF_INTERVAL_MINUTES[tf] * 60 + elapsed = (datetime.now(timezone.utc) - last_ts).total_seconds() + return elapsed < COOLDOWN_BARS * interval_s + + # ── Trade execution ──────────────────────────────────────────── + + def _execute_trade(self, report: SignalReport, tf: Timeframe) -> None: + ts = report.trade_setup + if ts is None: + return + + # All four TF loops can fire at an aligned boundary (e.g. :00) and + # pass _should_execute before any of them has recorded a fill — + # re-check under the entry lock so only one order goes out. + with self._entry_lock: + with self._state_lock: + if self._state.get("trades"): + logger.info( + "%s: position opened by a concurrent scan, skipping", + tf.value) + return + + # NO take-profit on the order: the backtested trailing model + # exits via SL/TRAIL/TIMEOUT only — a hard TP would cap + # winners and diverge the live book from the backtest. + try: + result = self._executor.place_market_order( + instrument=INSTRUMENT, + direction=ts.direction, + stop_loss=ts.stop_loss, + ) + except OandaAPIError: + logger.exception( + "Failed to place %s order for %s %s", + ts.direction, SYMBOL, tf.value) + return + + fill_tx = result.get("orderFillTransaction") + if not fill_tx: + logger.warning( + "Order not filled for %s %s", SYMBOL, tf.value) + return + + trade_opened = fill_tx.get("tradeOpened", {}) + trade_id = trade_opened.get("tradeID") + if not trade_id: + logger.warning("No trade ID in fill response") + return + + fill_price = float(fill_tx.get("price", ts.entry_price)) + risk = abs(fill_price - ts.stop_loss) + + trail_arm = ts.trail_arm_r if ts.trail_arm_r is not None else 0.6 + trail_dist = ts.trail_dist_r if ts.trail_dist_r is not None else 0.75 + + now_iso = datetime.now(timezone.utc).isoformat() + trade_state = { + "tf": tf.value, + "direction": ts.direction, + "entry": fill_price, + "sl_initial": ts.stop_loss, + "current_sl": ts.stop_loss, + "tp1": ts.take_profit_1, + "tp2": ts.take_profit_2, + "risk": risk, + "peak": fill_price, + "trail_armed": False, + "trail_arm_r": trail_arm, + "trail_dist_r": trail_dist, + "bars_held": 0, + # The signal bar is already closed; the first bar this + # trade lives through is the one that just started. Guard + # so a lagging feed can't trail on the signal bar itself. + "signal_bar_ts": report.timestamp, + "last_bar_ts": None, + "strategy": ts.strategy, + "quality": report.signal_quality, + "rr": ts.risk_reward, + "opened_at": now_iso, + } + + with self._state_lock: + self._state["trades"][trade_id] = trade_state + self._state["last_entries"][tf.value] = now_iso + _save_state(self._state) + + logger.info( + "TRADE OPENED: %s %s @ %.2f | SL %.2f | TP1 %.2f (advisory) | " + "risk $%.2f | RR %.2f | quality %s | strategy %s | TF %s | " + "trail arm %.2fR dist %.2fR", + ts.direction, SYMBOL, fill_price, ts.stop_loss, + ts.take_profit_1, + risk, ts.risk_reward, report.signal_quality, + ts.strategy, tf.value, trail_arm, trail_dist) + + # ── Trailing stop management ─────────────────────────────────── + + def _manage_trailing(self, tf: Timeframe) -> None: + with self._state_lock: + trades = dict(self._state.get("trades", {})) + + for trade_id, info in trades.items(): + if info.get("tf") != tf.value: + continue + + try: + df = self._provider.fetch_ohlcv(SYMBOL, tf) + except Exception: + logger.exception( + "Failed to fetch candle data for trail update") + continue + + if df.empty: + logger.warning( + "%s: no candle data for trail update of trade %s", + tf.value, trade_id) + continue + + bar_ts = df.index[-1].isoformat() + + # Process each closed bar exactly once. Weekends and feed lag + # re-serve the same last bar; without this guard bars_held + # inflates (an M5 trade would "time out" over a weekend) and + # the peak gets double-counted. + if bar_ts == info.get("last_bar_ts"): + continue + sig_ts = info.get("signal_bar_ts") + if sig_ts: + try: + if df.index[-1] <= pd.Timestamp(sig_ts): + continue + except (ValueError, TypeError): + pass + info["last_bar_ts"] = bar_ts + + info["bars_held"] = info.get("bars_held", 0) + 1 + + if info["bars_held"] >= self._max_hold_bars: + logger.info( + "TIMEOUT: trade %s held %d bars, closing", + trade_id, info["bars_held"]) + try: + 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. + logger.exception( + "Failed to close timed-out trade %s — will retry", + trade_id) + with self._state_lock: + self._state["trades"][trade_id] = info + _save_state(self._state) + continue + with self._state_lock: + self._state["trades"].pop(trade_id, None) + _save_state(self._state) + continue + + last_bar = df.iloc[-1] + bar_high = float(last_bar["High"]) + bar_low = float(last_bar["Low"]) + + new_sl = self._compute_trail(info, bar_high, bar_low) + + if new_sl is not None: + # Round to instrument precision and track exactly what the + # broker holds, so tracked vs broker SL can never drift. + new_sl = round(new_sl, self._price_precision) + if new_sl != info["current_sl"]: + try: + self._executor.modify_trade_sl(trade_id, new_sl) + info["current_sl"] = new_sl + except OandaAPIError: + logger.exception( + "Failed to update trailing SL for trade %s", + trade_id) + + with self._state_lock: + self._state["trades"][trade_id] = info + _save_state(self._state) + + def _compute_trail( + self, info: Dict, bar_high: float, bar_low: float + ) -> Optional[float]: + buy = info["direction"] == "BUY" + risk = info["risk"] + if risk <= 0: + return None + arm = info.get("trail_arm_r", 0.6) + dist = info.get("trail_dist_r", 0.75) + if arm <= 0: + return None + + fav = bar_high if buy else bar_low + if buy: + info["peak"] = max(info.get("peak", info["entry"]), fav) + else: + info["peak"] = min(info.get("peak", info["entry"]), fav) + + exc = ((info["peak"] - info["entry"]) if buy + else (info["entry"] - info["peak"])) / risk + + if exc >= arm: + if buy: + lvl = info["peak"] - dist * risk + cur_sl = info.get("current_sl", info["sl_initial"]) + if lvl > cur_sl: + info["trail_armed"] = True + return lvl + else: + lvl = info["peak"] + dist * risk + cur_sl = info.get("current_sl", info["sl_initial"]) + if lvl < cur_sl: + info["trail_armed"] = True + return lvl + + return None + + # ── Closed-trade detection ───────────────────────────────────── + + def _check_closed_trades(self) -> None: + with self._state_lock: + tracked = set(self._state.get("trades", {}).keys()) + + if not tracked: + return + + try: + open_trades = self._executor.get_open_trades(INSTRUMENT) + except OandaAPIError: + logger.exception("Failed to query open trades") + return + + live_ids = {t["id"] for t in open_trades} + closed = tracked - live_ids + + for tid in closed: + with self._state_lock: + info = self._state["trades"].pop(tid, {}) + entry = info.get("entry", 0) + direction = info.get("direction", "?") + risk = info.get("risk", 1) + sl = info.get("current_sl", 0) + was_trailed = info.get("trail_armed", False) + + logger.info( + "TRADE CLOSED BY BROKER: %s %s %s | entry %.2f | " + "SL was %.2f | risk $%.2f | trailed=%s | bars_held=%d", + tid, direction, SYMBOL, entry, sl, risk, + was_trailed, info.get("bars_held", 0)) + + if closed: + with self._state_lock: + _save_state(self._state) + +# ── Uncaught exception handlers ─────────────────────────────────────── + +def _excepthook(exc_type, exc_value, exc_tb): + if issubclass(exc_type, KeyboardInterrupt): + sys.__excepthook__(exc_type, exc_value, exc_tb) + return + logger.critical( + "Uncaught exception", exc_info=(exc_type, exc_value, exc_tb)) + +def _threading_excepthook(args): + logger.critical( + "Uncaught exception in thread %s", + args.thread.name if args.thread else "?", + exc_info=(args.exc_type, args.exc_value, args.exc_traceback)) + +# ── Entry point ──────────────────────────────────────────────────────── + +def main() -> None: + _setup_logging( + os.environ.get("OANDA_LOG_LEVEL") or get_app_config().log_level) + sys.excepthook = _excepthook + threading.excepthook = _threading_excepthook + + daemon = TradingDaemon() + asyncio.run(daemon.run()) + + +if __name__ == "__main__": + main() diff --git a/oanda_executor.py b/oanda_executor.py new file mode 100644 index 0000000..5a1a2d0 --- /dev/null +++ b/oanda_executor.py @@ -0,0 +1,212 @@ +""" +ForexBot2 — Oanda v20 Order Executor +====================================== +Handles all Oanda account operations: market orders, trade queries, +SL/TP modification, and trade closure. Uses raw ``requests`` against +the Oanda v20 REST API — no extra dependencies. +""" + +import logging +import time +from typing import Dict, List, Optional + +import requests + +logger = logging.getLogger(__name__) + + +class OandaAPIError(Exception): + """Raised when an Oanda API call fails.""" + + def __init__(self, message: str, status_code: int = 0, body: str = ""): + super().__init__(message) + self.status_code = status_code + self.body = body + + +class OandaExecutor: + """Oanda v20 order execution and trade management.""" + + MAX_RETRIES = 1 + RETRY_DELAY = 2.0 + + def __init__( + self, + account_id: str, + api_token: str, + base_url: str = "https://api-fxpractice.oanda.com", + lot_size: int = 1, + price_precision: int = 2, + ): + self._account_id = account_id + self._base_url = base_url.rstrip("/") + self._lot_size = lot_size + self._price_precision = price_precision + self._session = requests.Session() + self._session.headers.update({ + "Authorization": f"Bearer {api_token}", + "Accept-Datetime-Format": "RFC3339", + "Content-Type": "application/json", + }) + + # ── Order Placement ──────────────────────────────────────────── + + def place_market_order( + self, + instrument: str, + direction: str, + stop_loss: float, + take_profit: Optional[float] = None, + units: Optional[int] = None, + ) -> Dict: + if units is None: + units = self._lot_size + signed_units = str(units) if direction.upper() == "BUY" else str(-units) + + order_body = { + "order": { + "type": "MARKET", + "instrument": instrument, + "units": signed_units, + "timeInForce": "FOK", + "stopLossOnFill": { + "price": self._fmt_price(stop_loss), + }, + } + } + if take_profit is not None: + order_body["order"]["takeProfitOnFill"] = { + "price": self._fmt_price(take_profit), + } + + url = f"{self._base_url}/v3/accounts/{self._account_id}/orders" + # Never retry order placement: a timeout after the server already + # filled would open a SECOND position on retry. Fail loudly instead; + # the next candle boundary re-evaluates. + resp = self._request_with_retry("POST", url, json=order_body, retries=0) + data = self._parse_json(resp) + + fill_tx = data.get("orderFillTransaction") + if fill_tx: + trade_id = fill_tx.get("tradeOpened", {}).get("tradeID", "unknown") + fill_price = float(fill_tx.get("price", 0)) + logger.info( + "ORDER FILLED: %s %s %s units @ %.2f — trade %s, SL %.2f%s", + direction.upper(), instrument, signed_units, fill_price, + trade_id, stop_loss, + f", TP {take_profit:.2f}" if take_profit else "") + else: + cancel_tx = data.get("orderCancelTransaction", {}) + reason = cancel_tx.get("reason", "unknown") + logger.warning( + "ORDER NOT FILLED: %s %s — reason: %s", + direction.upper(), instrument, reason) + + return data + + # ── Trade Queries ────────────────────────────────────────────── + + def get_open_trades(self, instrument: Optional[str] = None) -> List[Dict]: + url = f"{self._base_url}/v3/accounts/{self._account_id}/openTrades" + resp = self._request_with_retry("GET", url) + trades = self._parse_json(resp).get("trades", []) + if instrument: + trades = [t for t in trades if t.get("instrument") == instrument] + return trades + + def get_trade(self, trade_id: str) -> Dict: + url = (f"{self._base_url}/v3/accounts/{self._account_id}" + f"/trades/{trade_id}") + resp = self._request_with_retry("GET", url) + return self._parse_json(resp).get("trade", {}) + + def get_account_summary(self) -> Dict: + url = f"{self._base_url}/v3/accounts/{self._account_id}/summary" + resp = self._request_with_retry("GET", url) + account = self._parse_json(resp).get("account", {}) + logger.info( + "Account: balance=%.2f, NAV=%.2f, unrealizedPL=%.2f, openTrades=%s", + float(account.get("balance", 0)), + float(account.get("NAV", 0)), + float(account.get("unrealizedPL", 0)), + account.get("openTradeCount", 0)) + return account + + # ── Trade Modification ───────────────────────────────────────── + + def modify_trade_sl(self, trade_id: str, new_sl: float) -> Dict: + url = (f"{self._base_url}/v3/accounts/{self._account_id}" + f"/trades/{trade_id}/orders") + body = {"stopLoss": {"price": self._fmt_price(new_sl)}} + resp = self._request_with_retry("PUT", url, json=body) + logger.info( + "TRAIL UPDATE: trade %s — new SL %.2f", trade_id, new_sl) + return self._parse_json(resp) + + def modify_trade_tp(self, trade_id: str, new_tp: float) -> Dict: + url = (f"{self._base_url}/v3/accounts/{self._account_id}" + f"/trades/{trade_id}/orders") + body = {"takeProfit": {"price": self._fmt_price(new_tp)}} + resp = self._request_with_retry("PUT", url, json=body) + logger.info("TP UPDATE: trade %s — new TP %.2f", trade_id, new_tp) + return self._parse_json(resp) + + def close_trade(self, trade_id: str) -> Dict: + url = (f"{self._base_url}/v3/accounts/{self._account_id}" + 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_price = float(close_tx.get("price", 0)) + pl = float(close_tx.get("pl", 0)) + logger.info( + "TRADE CLOSED: trade %s @ %.2f — PL %.2f", + trade_id, close_price, pl) + return data + + # ── Internals ────────────────────────────────────────────────── + + @staticmethod + def _parse_json(resp: requests.Response) -> Dict: + try: + return resp.json() + except ValueError as e: + raise OandaAPIError( + f"Oanda returned a non-JSON response (HTTP {resp.status_code})", + resp.status_code, resp.text[:300]) from e + + def _fmt_price(self, price: float) -> str: + # Oanda rejects prices with more decimals than the instrument's + # displayPrecision (XAU_USD: sub-cent precision is invalid). + return f"{price:.{self._price_precision}f}" + + def _request_with_retry( + self, method: str, url: str, retries: Optional[int] = None, **kwargs + ) -> requests.Response: + if retries is None: + retries = self.MAX_RETRIES + last_err = None + for attempt in range(1 + retries): + try: + resp = self._session.request(method, url, timeout=15, **kwargs) + if resp.status_code >= 500: + raise OandaAPIError( + f"Oanda server error {resp.status_code}", + resp.status_code, resp.text) + if resp.status_code >= 400: + raise OandaAPIError( + f"Oanda API error {resp.status_code}: {resp.text}", + resp.status_code, resp.text) + return resp + except (requests.ConnectionError, requests.Timeout, + OandaAPIError) as e: + last_err = e + if isinstance(e, OandaAPIError) and 400 <= e.status_code < 500: + raise + if attempt < retries: + logger.warning( + "Oanda request failed (attempt %d/%d): %s — retrying in %.0fs", + attempt + 1, 1 + retries, e, + self.RETRY_DELAY) + time.sleep(self.RETRY_DELAY) + raise OandaAPIError(f"Oanda request failed after retries: {last_err}") diff --git a/oanda_provider.py b/oanda_provider.py new file mode 100644 index 0000000..c421fda --- /dev/null +++ b/oanda_provider.py @@ -0,0 +1,361 @@ +""" +ForexBot2 — Oanda v20 Data Provider +==================================== +Drop-in replacement for ForexDataProvider that fetches OHLCV candles +from the Oanda v20 REST API. Returns the same DataFrame format +(Open/High/Low/Close/Volume, UTC DatetimeIndex, oldest-first, closed +bars only) so the signal pipeline (engine + aggregator) runs unchanged. + +Additionally exposes real bid/ask spread from Oanda's MBA candles. +""" + +import logging +import threading +import time +from typing import Dict, Optional + +import pandas as pd +import requests + +from config import PAIR_INFO, PlatformConfig, Timeframe, XAUUSD + +logger = logging.getLogger(__name__) + + +class OandaDataError(Exception): + """Raised when an Oanda data fetch fails or returns invalid data.""" + + +def _is_forex_weekend(now=None) -> bool: + 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: + return True + return False + + +class OandaDataProvider: + """Oanda v20 REST candle provider with in-memory caching.""" + + OANDA_GRANULARITY = { + Timeframe.M1: "M1", + Timeframe.M5: "M5", + Timeframe.M15: "M15", + Timeframe.M30: "M30", + Timeframe.H1: "H1", + Timeframe.H4: "H4", + Timeframe.D1: "D", + Timeframe.W1: "W", + } + + OANDA_INSTRUMENT = { + "XAU/USD": "XAU_USD", + } + + BARS_PER_DAY = { + Timeframe.M1: 1440, Timeframe.M5: 288, Timeframe.M15: 96, + Timeframe.M30: 48, Timeframe.H1: 24, Timeframe.H4: 6, + Timeframe.D1: 1, Timeframe.W1: 0.143, + } + + CANDLE_SECONDS = { + Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900, + Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, + Timeframe.D1: 86400, Timeframe.W1: 604800, + } + + FRESH_FETCH_RETRY_SECONDS = 3.0 + + def __init__(self, config: PlatformConfig, api_token: str, base_url: str): + self.config = config + self._base_url = base_url.rstrip("/") + self._session = requests.Session() + self._session.headers.update({ + "Authorization": f"Bearer {api_token}", + "Accept-Datetime-Format": "RFC3339", + "Content-Type": "application/json", + }) + self._cache: Dict[str, pd.DataFrame] = {} + self._cache_timestamps: Dict[str, float] = {} + self._cache_lock = threading.Lock() + self._spread_cache: Dict[str, float] = {} + + # ── Public API ───────────────────────────────────────────────── + + def fetch_ohlcv( + self, + symbol: str = XAUUSD, + timeframe: Timeframe = Timeframe.H1, + lookback_days: Optional[int] = None, + ) -> pd.DataFrame: + if lookback_days is None: + lookback_days = self.config.data_lookback_days + + cache_key = f"{symbol}_{timeframe.value}" + now = time.time() + + cache_ttl = self.CANDLE_SECONDS.get(timeframe, 3600) + with self._cache_lock: + cached = self._cache.get(cache_key) + cached_at = self._cache_timestamps.get(cache_key) + if ( + cached is not None + and cached_at is not None + and (now - cached_at) < cache_ttl + and not cached.empty + ): + if (self._is_data_fresh(cached, symbol, timeframe) + and self._has_latest_closed_bar(cached, timeframe)): + return cached + logger.warning( + "%s %s: cached data is stale, refetching", + symbol, timeframe.value) + with self._cache_lock: + self._cache.pop(cache_key, None) + self._cache_timestamps.pop(cache_key, None) + + count = max(300, min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000)) + df = self._fetch_oanda_candles(symbol, timeframe, count) + df = self._validate_and_clean(df, symbol, timeframe) + + if not df.empty and not self._has_latest_closed_bar(df, timeframe): + time.sleep(self.FRESH_FETCH_RETRY_SECONDS) + try: + retry = self._fetch_oanda_candles(symbol, timeframe, count) + retry = self._validate_and_clean(retry, symbol, timeframe) + if retry is not None and not retry.empty: + df = retry + except OandaDataError as e: + logger.warning( + "%s %s: boundary retry refetch failed (%s) — keeping " + "original data", symbol, timeframe.value, e) + if not self._has_latest_closed_bar(df, timeframe): + logger.warning( + "%s %s: feed lagging — just-closed bar still missing " + "after retry; this scan may run one bar behind", + symbol, timeframe.value) + + if not self._is_data_fresh(df, symbol, timeframe): + logger.warning( + "STALE DATA from Oanda: %s %s — last bar %s is too old. " + "Signal quality may be degraded.", + symbol, timeframe.value, + df.index[-1] if not df.empty else "N/A") + + with self._cache_lock: + self._cache[cache_key] = df + self._cache_timestamps[cache_key] = now + return df + + def get_pair_info(self, symbol: str = XAUUSD) -> Dict: + info = PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD]) + return {"symbol": symbol, **info} + + def get_spread(self, symbol: str = XAUUSD) -> float: + return self._spread_cache.get( + symbol, PAIR_INFO.get(symbol, PAIR_INFO[XAUUSD])["default_spread"]) + + def clear_cache(self) -> int: + with self._cache_lock: + count = len(self._cache) + self._cache.clear() + self._cache_timestamps.clear() + logger.info("Cache cleared: %d entries wiped", count) + return count + + # ── Private ──────────────────────────────────────────────────── + + def _to_oanda_instrument(self, symbol: str) -> str: + if symbol in self.OANDA_INSTRUMENT: + return self.OANDA_INSTRUMENT[symbol] + return symbol.replace("/", "_") + + def _fetch_oanda_candles( + self, symbol: str, timeframe: Timeframe, count: int + ) -> pd.DataFrame: + instrument = self._to_oanda_instrument(symbol) + granularity = self.OANDA_GRANULARITY.get(timeframe) + if granularity is None: + raise OandaDataError( + f"Unsupported timeframe {timeframe} for Oanda") + + url = f"{self._base_url}/v3/instruments/{instrument}/candles" + params = { + "granularity": granularity, + "count": min(count, 5000), + "price": "MBA", + "dailyAlignment": 0, + "alignmentTimezone": "Etc/UTC", + } + + try: + resp = self._session.get(url, params=params, timeout=30) + except requests.RequestException as e: + raise OandaDataError( + f"Oanda candle fetch failed for {instrument}: {e}") from e + + if resp.status_code != 200: + raise OandaDataError( + f"Oanda candle fetch failed for {instrument}: " + f"HTTP {resp.status_code}: {resp.text[:300]}") + + try: + data = resp.json() + except ValueError as e: + raise OandaDataError( + f"Oanda candle response for {instrument} is not valid " + f"JSON: {e}") from e + candles = data.get("candles", []) + if not candles: + raise OandaDataError( + f"Empty candle response from Oanda for {instrument}") + + rows = [] + last_bid_close = None + last_ask_close = None + for c in candles: + if not c.get("complete", False): + continue + mid = c.get("mid", {}) + bid = c.get("bid", {}) + ask = c.get("ask", {}) + 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)), + "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 not rows: + raise OandaDataError( + f"No complete candles from Oanda for {instrument}") + + if last_bid_close and last_ask_close: + self._spread_cache[symbol] = round( + last_ask_close - last_bid_close, 4) + + df = pd.DataFrame(rows) + df = df.set_index("Datetime") + df = df.sort_index() + if df.index.tz is None: + df.index = df.index.tz_localize("UTC") + df.index.name = "Datetime" + + for col in ("Open", "High", "Low", "Close"): + df[col] = pd.to_numeric(df[col], errors="coerce") + df["Volume"] = pd.to_numeric(df["Volume"], errors="coerce").fillna(0).astype(float) + + logger.debug( + "Fetched %d bars for %s @ %s from Oanda", + len(df), instrument, granularity) + return df + + def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool: + if df.empty: + return False + now = pd.Timestamp.now(tz="UTC") + if _is_forex_weekend(now): + return True + candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) + dur = pd.Timedelta(seconds=candle_seconds) + expected_label = now.floor(dur) - dur + return df.index[-1] >= expected_label + + def _is_data_fresh( + self, df: pd.DataFrame, symbol: str, timeframe: Timeframe + ) -> bool: + if df is None or df.empty: + return False + now = pd.Timestamp.now(tz="UTC") + if _is_forex_weekend(now): + return True + last_bar = df.index[-1] + candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) + max_age_seconds = candle_seconds * 3 + age_seconds = (now - last_bar).total_seconds() + if age_seconds > max_age_seconds: + logger.warning( + "%s %s: last bar %s is %dm old (limit: %dm)", + symbol, timeframe.value, + last_bar.strftime("%Y-%m-%d %H:%M"), + age_seconds / 60, max_age_seconds / 60) + return False + return True + + def _validate_and_clean( + self, df: pd.DataFrame, symbol: str, timeframe: Timeframe + ) -> pd.DataFrame: + if df is None or df.empty: + raise OandaDataError(f"Empty dataset for {symbol}") + + original_len = len(df) + + df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all") + df = df.ffill(limit=3) + df = df.dropna(subset=["Close"]) + + df = df.sort_index() + df = df[~df.index.duplicated(keep="last")] + + now = pd.Timestamp.utcnow() + future_mask = df.index > now + pd.Timedelta(minutes=5) + if future_mask.any(): + logger.warning( + "%s: Dropping %d future timestamps", symbol, future_mask.sum()) + df = df[~future_mask] + + invalid_ohlc = ( + (df["High"] < df["Low"]) + | (df["High"] < df["Open"]) + | (df["High"] < df["Close"]) + | (df["Low"] > df["Open"]) + | (df["Low"] > df["Close"]) + ) + if invalid_ohlc.any(): + logger.warning( + "%s: Fixing %d OHLC-inconsistent bars", + symbol, invalid_ohlc.sum()) + for idx in df[invalid_ohlc].index: + vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]] + df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]]) + df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) + + returns = df["Close"].pct_change() + std_ret = returns.std() + if std_ret > 0: + dev = returns - returns.mean() + outlier_mask = dev.abs() > 10 * std_ret + recovery = ( + outlier_mask + & outlier_mask.shift(1, fill_value=False) + & (dev * dev.shift(1) < 0) + ) + outlier_mask = outlier_mask & ~recovery + if outlier_mask.any(): + logger.warning( + "%s: Removing %d extreme outlier bars", + symbol, outlier_mask.sum()) + df = df[~outlier_mask] + + min_required = max(self.config.technical.ema_trend, 200) + 10 + if len(df) < min_required: + logger.warning( + "%s %s: Only %d bars (need %d). Some indicators will be partial.", + symbol, timeframe.value, len(df), min_required) + + cleaned = original_len - len(df) + if cleaned > 0: + logger.info("%s: Cleaned %d/%d bars", symbol, cleaned, original_len) + + return df diff --git a/test_all.py b/test_all.py index b651b8d..438f97b 100644 --- a/test_all.py +++ b/test_all.py @@ -8,6 +8,7 @@ Tests every function in the codebase without requiring live API connections. import sys import os import time +import threading import traceback from datetime import datetime, timedelta, timezone from unittest.mock import patch, MagicMock @@ -3121,6 +3122,847 @@ def _(): tb._autoscan_chats.clear() +# ══════════════════════════════════════════════════════════════════════ +# MODULE 15: Oanda provider / executor / daemon +# ══════════════════════════════════════════════════════════════════════ + +print("\n═══ Module 15: Oanda provider / executor / daemon ═══") + +import oanda_provider +import oanda_executor +import oanda_daemon + + +@test("oanda_provider: _is_forex_weekend matches data_provider's") +def _(): + from data_provider import _is_forex_weekend as dp_weekend + from oanda_provider import _is_forex_weekend as oa_weekend + import pandas as pd + for ts_str in [ + "2026-07-17 12:00", # Friday noon + "2026-07-17 22:30", # Friday after close + "2026-07-18 15:00", # Saturday + "2026-07-19 12:00", # Sunday before open + "2026-07-19 22:30", # Sunday after open + "2026-07-20 08:00", # Monday + ]: + ts = pd.Timestamp(ts_str, tz="UTC") + assert dp_weekend(ts) == oa_weekend(ts), \ + f"Weekend check mismatch at {ts_str}" + + +@test("oanda_provider: OandaDataProvider parses Oanda candle JSON into correct DataFrame") +def _(): + sample_candles = [ + { + "complete": True, + "time": "2026-07-20T10:00:00.000000000Z", + "mid": {"o": "2400.00", "h": "2405.00", "l": "2398.00", "c": "2403.00"}, + "bid": {"o": "2399.85", "h": "2404.85", "l": "2397.85", "c": "2402.85"}, + "ask": {"o": "2400.15", "h": "2405.15", "l": "2398.15", "c": "2403.15"}, + "volume": 1234, + }, + { + "complete": True, + "time": "2026-07-20T11:00:00.000000000Z", + "mid": {"o": "2403.00", "h": "2408.00", "l": "2401.00", "c": "2406.00"}, + "bid": {"o": "2402.85", "h": "2407.85", "l": "2400.85", "c": "2405.85"}, + "ask": {"o": "2403.15", "h": "2408.15", "l": "2401.15", "c": "2406.15"}, + "volume": 5678, + }, + { + "complete": False, + "time": "2026-07-20T12:00:00.000000000Z", + "mid": {"o": "2406.00", "h": "2406.50", "l": "2405.50", "c": "2406.20"}, + "bid": {"o": "2405.85", "h": "2406.35", "l": "2405.35", "c": "2406.05"}, + "ask": {"o": "2406.15", "h": "2406.65", "l": "2405.65", "c": "2406.35"}, + "volume": 100, + }, + ] + + mock_resp = type("MockResp", (), { + "status_code": 200, + "json": lambda self: {"candles": sample_candles, "instrument": "XAU_USD"}, + "raise_for_status": lambda self: None, + })() + + provider2 = oanda_provider.OandaDataProvider.__new__( + oanda_provider.OandaDataProvider) + provider2.config = PlatformConfig() + provider2._base_url = "https://test" + provider2._cache = {} + provider2._cache_timestamps = {} + provider2._cache_lock = threading.Lock() + provider2._spread_cache = {} + provider2._session = type("S", (), { + "get": lambda self, *a, **k: mock_resp, + "headers": {}, + })() + + df = provider2._fetch_oanda_candles("XAU/USD", Timeframe.H1, 300) + + assert len(df) == 2, f"Expected 2 complete bars, got {len(df)}" + assert list(df.columns) == ["Open", "High", "Low", "Close", "Volume"], \ + f"Wrong columns: {list(df.columns)}" + assert df.index.name == "Datetime" + assert df.index.tz is not None, "Index must be tz-aware" + assert df.iloc[0]["Open"] == 2400.0 + assert df.iloc[1]["Close"] == 2406.0 + assert df.iloc[0]["Volume"] == 1234 + + spread = provider2._spread_cache.get("XAU/USD") + assert spread is not None, "Spread should be cached" + assert abs(spread - 0.30) < 0.01, f"Expected spread ~0.30, got {spread}" + + +@test("oanda_provider: instrument name mapping") +def _(): + provider = oanda_provider.OandaDataProvider.__new__( + oanda_provider.OandaDataProvider) + assert provider._to_oanda_instrument("XAU/USD") == "XAU_USD" + assert provider._to_oanda_instrument("EUR/USD") == "EUR_USD" + + +@test("oanda_provider: granularity mapping covers all scan timeframes") +def _(): + for tf in [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]: + assert tf in oanda_provider.OandaDataProvider.OANDA_GRANULARITY, \ + f"{tf} missing from OANDA_GRANULARITY" + + +@test("oanda_provider: get_spread returns default when no cache") +def _(): + provider = oanda_provider.OandaDataProvider.__new__( + oanda_provider.OandaDataProvider) + provider._spread_cache = {} + spread = provider.get_spread("XAU/USD") + assert spread == 0.30, f"Expected default spread 0.30, got {spread}" + + +@test("oanda_executor: market order body construction") +def _(): + executor = oanda_executor.OandaExecutor.__new__( + oanda_executor.OandaExecutor) + executor._account_id = "test-account" + executor._base_url = "https://test" + executor._lot_size = 1 + executor._price_precision = 2 + executor._session = None + + captured = {} + + def mock_request(self, method, url, **kwargs): + captured["method"] = method + captured["url"] = url + captured["json"] = kwargs.get("json") + return type("R", (), { + "status_code": 201, + "json": lambda self: { + "orderFillTransaction": { + "price": "2400.50", + "tradeOpened": {"tradeID": "123"}, + } + }, + })() + + executor._session = type("S", (), { + "request": mock_request, + "headers": {}, + })() + + result = executor.place_market_order( + instrument="XAU_USD", + direction="BUY", + stop_loss=2392.50, + take_profit=2420.00, + ) + + body = captured["json"] + order = body["order"] + assert order["type"] == "MARKET" + assert order["instrument"] == "XAU_USD" + assert order["units"] == "1" + # XAU price precision is 2 — Oanda rejects excess decimals. + assert order["stopLossOnFill"]["price"] == "2392.50" + assert order["takeProfitOnFill"]["price"] == "2420.00" + + fill = result.get("orderFillTransaction") + assert fill is not None + assert fill["tradeOpened"]["tradeID"] == "123" + + +@test("oanda_executor: SELL order uses negative units") +def _(): + executor = oanda_executor.OandaExecutor.__new__( + oanda_executor.OandaExecutor) + executor._account_id = "test-account" + executor._base_url = "https://test" + executor._lot_size = 1 + executor._price_precision = 2 + + captured = {} + + def mock_request(self, method, url, **kwargs): + captured["json"] = kwargs.get("json") + return type("R", (), { + "status_code": 201, + "json": lambda self: { + "orderFillTransaction": { + "price": "2400.50", + "tradeOpened": {"tradeID": "456"}, + } + }, + })() + + executor._session = type("S", (), { + "request": mock_request, + "headers": {}, + })() + + executor.place_market_order( + instrument="XAU_USD", direction="SELL", stop_loss=2410.00) + + order = captured["json"]["order"] + assert order["units"] == "-1", f"SELL should be -1, got {order['units']}" + assert "takeProfitOnFill" not in order, "No TP should be set when None" + + +@test("oanda_executor: 4xx error raises immediately (no retry)") +def _(): + executor = oanda_executor.OandaExecutor.__new__( + oanda_executor.OandaExecutor) + executor._account_id = "test" + executor._base_url = "https://test" + executor._lot_size = 1 + call_count = {"n": 0} + + def mock_request(self, method, url, **kwargs): + call_count["n"] += 1 + return type("R", (), { + "status_code": 400, + "text": "INVALID_ORDER", + })() + + executor._session = type("S", (), { + "request": mock_request, + "headers": {}, + })() + + try: + executor._request_with_retry("POST", "https://test/orders") + assert False, "Should have raised OandaAPIError" + except oanda_executor.OandaAPIError as e: + assert e.status_code == 400 + assert call_count["n"] == 1, "4xx should not retry" + + +@test("oanda_daemon: trailing stop ratchet matches backtester logic") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + + info_buy = { + "direction": "BUY", + "entry": 2400.0, + "sl_initial": 2384.0, + "current_sl": 2384.0, + "risk": 16.0, + "peak": 2400.0, + "trail_armed": False, + "trail_arm_r": 0.6, + "trail_dist_r": 0.75, + } + + # Bar 1: price moves up but not enough to arm (< 0.6R = $9.60) + new_sl = daemon._compute_trail(info_buy, bar_high=2408.0, bar_low=2399.0) + assert new_sl is None, "Should not arm yet (excursion 0.5R)" + assert info_buy["peak"] == 2408.0 + + # Bar 2: excursion reaches 0.625R = $10 — arms, trail = 2408 - 0.75*16 = 2396 + new_sl = daemon._compute_trail(info_buy, bar_high=2410.0, bar_low=2403.0) + assert new_sl is not None, "Should arm at 0.625R" + assert abs(new_sl - 2398.0) < 0.01, f"Expected trail ~2398, got {new_sl}" + assert info_buy["trail_armed"] is True + + # Bar 3: price goes higher, trail ratchets up + info_buy["current_sl"] = new_sl + new_sl2 = daemon._compute_trail(info_buy, bar_high=2420.0, bar_low=2410.0) + assert new_sl2 is not None + assert new_sl2 > new_sl, "Trail must ratchet up" + expected = 2420.0 - 0.75 * 16.0 # 2408.0 + assert abs(new_sl2 - expected) < 0.01, f"Expected {expected}, got {new_sl2}" + + # Bar 4: price pulls back — trail must NOT ratchet down + info_buy["current_sl"] = new_sl2 + new_sl3 = daemon._compute_trail(info_buy, bar_high=2412.0, bar_low=2405.0) + assert new_sl3 is None, "Trail must not ratchet backwards" + + +@test("oanda_daemon: trailing stop for SELL direction") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + + info_sell = { + "direction": "SELL", + "entry": 2400.0, + "sl_initial": 2416.0, + "current_sl": 2416.0, + "risk": 16.0, + "peak": 2400.0, + "trail_armed": False, + "trail_arm_r": 0.6, + "trail_dist_r": 0.75, + } + + # Not enough excursion + new_sl = daemon._compute_trail(info_sell, bar_high=2401.0, bar_low=2393.0) + assert new_sl is None, "Should not arm yet" + assert info_sell["peak"] == 2393.0 + + # Arms: peak at 2390, excursion = 10/16 = 0.625R + new_sl = daemon._compute_trail(info_sell, bar_high=2395.0, bar_low=2390.0) + assert new_sl is not None + expected = 2390.0 + 0.75 * 16.0 # 2402.0 + assert abs(new_sl - expected) < 0.01, f"Expected {expected}, got {new_sl}" + assert info_sell["trail_armed"] is True + + # Ratchets down + info_sell["current_sl"] = new_sl + new_sl2 = daemon._compute_trail(info_sell, bar_high=2392.0, bar_low=2380.0) + assert new_sl2 is not None + assert new_sl2 < new_sl, "SELL trail must ratchet down" + + +@test("oanda_daemon: _should_execute respects quality gate") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._quality_gate = "B" + daemon._state = {"trades": {}, "last_entries": {}} + daemon._state_lock = threading.Lock() + + from signal_aggregator import TradeSetup + setup = TradeSetup( + pair="XAU/USD", display_name="XAU/USD", direction="BUY", + entry_price=2400.0, stop_loss=2384.0, + take_profit_1=2416.0, take_profit_2=2432.0, + risk_reward=1.0, risk_pips=160.0, reward_pips=160.0, + pip_value=0.10) + + report_c = SignalReport( + symbol="XAU/USD", display_name="XAU/USD", timestamp="2026-07-20") + report_c.final_signal = SignalStrength.BUY + report_c.signal_quality = "C" + report_c.trade_setup = setup + + assert not daemon._should_execute(report_c, Timeframe.H1), \ + "C should be blocked by gate=B" + + report_a = SignalReport( + symbol="XAU/USD", display_name="XAU/USD", timestamp="2026-07-20") + report_a.final_signal = SignalStrength.BUY + report_a.signal_quality = "A" + report_a.trade_setup = setup + + assert daemon._should_execute(report_a, Timeframe.H1), \ + "A should pass gate=B" + + +@test("oanda_daemon: _should_execute blocks when position exists") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._quality_gate = "C" + daemon._state = {"trades": {"12345": {"direction": "BUY"}}, + "last_entries": {}} + daemon._state_lock = threading.Lock() + + from signal_aggregator import TradeSetup + setup = TradeSetup( + pair="XAU/USD", display_name="XAU/USD", direction="BUY", + entry_price=2400.0, stop_loss=2384.0, + take_profit_1=2416.0, take_profit_2=2432.0, + risk_reward=1.0, risk_pips=160.0, reward_pips=160.0, + pip_value=0.10) + + report = SignalReport( + symbol="XAU/USD", display_name="XAU/USD", timestamp="2026-07-20") + report.final_signal = SignalStrength.BUY + report.signal_quality = "A" + report.trade_setup = setup + + assert not daemon._should_execute(report, Timeframe.H1), \ + "Should block when a position already exists" + + +@test("oanda_daemon: NEUTRAL signal is never executed") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._quality_gate = "C" + daemon._state = {"trades": {}, "last_entries": {}} + daemon._state_lock = threading.Lock() + + report = SignalReport( + symbol="XAU/USD", display_name="XAU/USD", timestamp="2026-07-20") + report.final_signal = SignalStrength.NEUTRAL + report.signal_quality = "F" + report.trade_setup = None + + assert not daemon._should_execute(report, Timeframe.H1) + + +def _make_daemon_report(quality="A", direction="BUY"): + from signal_aggregator import TradeSetup + setup = TradeSetup( + pair="XAU/USD", display_name="XAU/USD", direction=direction, + entry_price=2400.0, stop_loss=2384.0, + take_profit_1=2416.0, take_profit_2=2432.0, + risk_reward=1.0, risk_pips=160.0, reward_pips=160.0, + pip_value=0.10, trail_arm_r=0.6, trail_dist_r=0.85) + report = SignalReport( + symbol="XAU/USD", display_name="XAU/USD", + timestamp="2026-07-20 10:00:00+00:00") + report.final_signal = (SignalStrength.BUY if direction == "BUY" + else SignalStrength.SELL) + report.signal_quality = quality + report.trade_setup = setup + return report + + +@test("oanda_daemon: 5-bar entry cooldown blocks re-entry (backtest parity)") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._quality_gate = "C" + daemon._state = {"trades": {}, "last_entries": {}} + daemon._state_lock = threading.Lock() + + report = _make_daemon_report() + + # Entry 2 minutes ago on M5 → inside 5×5min cooldown → blocked + recent = (datetime.now(timezone.utc) - timedelta(minutes=2)).isoformat() + daemon._state["last_entries"]["5min"] = recent + assert not daemon._should_execute(report, Timeframe.M5), \ + "Entry 2min ago must block M5 (cooldown 25min)" + + # Entry 30 minutes ago on M5 → past 5×5min cooldown → allowed + old = (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat() + daemon._state["last_entries"]["5min"] = old + assert daemon._should_execute(report, Timeframe.M5), \ + "Entry 30min ago must clear the 25min M5 cooldown" + + # M5 cooldown must not block H1 + daemon._state["last_entries"] = {"5min": recent} + assert daemon._should_execute(report, Timeframe.H1), \ + "Cooldown is per-timeframe" + + +@test("oanda_daemon: orders carry NO take-profit (trailing book parity)") +def _(): + import tempfile + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._state = {"trades": {}, "last_entries": {}} + daemon._state_lock = threading.Lock() + daemon._entry_lock = threading.Lock() + + captured = {} + + class MockExecutor: + def place_market_order(self, instrument, direction, stop_loss, + take_profit=None, units=None): + captured["take_profit"] = take_profit + captured["stop_loss"] = stop_loss + return { + "orderFillTransaction": { + "price": "2400.60", + "tradeOpened": {"tradeID": "777"}, + } + } + + daemon._executor = MockExecutor() + + with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: + tmp_path = f.name + orig_state_file = oanda_daemon.STATE_FILE + try: + oanda_daemon.STATE_FILE = tmp_path + daemon._execute_trade(_make_daemon_report(), Timeframe.M15) + finally: + oanda_daemon.STATE_FILE = orig_state_file + os.unlink(tmp_path) + + assert captured.get("take_profit") is None, \ + "Backtested trailing model has no hard TP — order must not carry one" + assert captured["stop_loss"] == 2384.0 + trade = daemon._state["trades"]["777"] + assert trade["entry"] == 2400.60, "Risk must be computed from actual fill" + assert abs(trade["risk"] - 16.60) < 0.01 + assert trade["trail_arm_r"] == 0.6 and trade["trail_dist_r"] == 0.85, \ + "Trail params must come from the setup stamp (per-TF factory values)" + assert daemon._state["last_entries"]["15min"], \ + "Entry must be recorded for the cooldown" + + +@test("oanda_daemon: entry-lock recheck prevents double entry from concurrent scans") +def _(): + import tempfile + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + # A concurrent TF loop already opened a trade after our _should_execute + daemon._state = {"trades": {"111": {"direction": "BUY"}}, + "last_entries": {}} + daemon._state_lock = threading.Lock() + daemon._entry_lock = threading.Lock() + + placed = {"n": 0} + + class MockExecutor: + def place_market_order(self, *a, **k): + placed["n"] += 1 + return {} + + daemon._executor = MockExecutor() + + with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: + tmp_path = f.name + orig_state_file = oanda_daemon.STATE_FILE + try: + oanda_daemon.STATE_FILE = tmp_path + daemon._execute_trade(_make_daemon_report(), Timeframe.M5) + finally: + oanda_daemon.STATE_FILE = orig_state_file + os.unlink(tmp_path) + + assert placed["n"] == 0, \ + "Recheck under the entry lock must skip when a position exists" + + +@test("oanda_daemon: same closed bar is never processed twice") +def _(): + import tempfile + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._max_hold_bars = 300 + daemon._price_precision = 2 + daemon._state_lock = threading.Lock() + + bar_time = pd.Timestamp("2026-07-20 10:00:00", tz="UTC") + df = pd.DataFrame( + {"Open": [2400.0], "High": [2405.0], "Low": [2398.0], + "Close": [2403.0], "Volume": [100.0]}, + index=pd.DatetimeIndex([bar_time], name="Datetime")) + + class MockProvider: + def fetch_ohlcv(self, symbol, tf): + return df + + daemon._provider = MockProvider() + daemon._executor = None # trail must not arm on this bar + + daemon._state = { + "trades": { + "555": { + "tf": "5min", "direction": "BUY", "entry": 2400.0, + "sl_initial": 2384.0, "current_sl": 2384.0, "risk": 16.0, + "peak": 2400.0, "trail_armed": False, + "trail_arm_r": 0.6, "trail_dist_r": 0.85, + "bars_held": 0, "last_bar_ts": None, "signal_bar_ts": None, + } + }, + "last_entries": {}, + } + + with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: + tmp_path = f.name + orig_state_file = oanda_daemon.STATE_FILE + try: + oanda_daemon.STATE_FILE = tmp_path + daemon._manage_trailing(Timeframe.M5) + daemon._manage_trailing(Timeframe.M5) # same bar re-served (weekend/lag) + daemon._manage_trailing(Timeframe.M5) + finally: + oanda_daemon.STATE_FILE = orig_state_file + os.unlink(tmp_path) + + assert daemon._state["trades"]["555"]["bars_held"] == 1, \ + "A re-served bar must not inflate bars_held (weekend timeout bug)" + + +@test("oanda_daemon: failed timeout-close keeps the trade tracked") +def _(): + import tempfile + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._max_hold_bars = 1 + daemon._price_precision = 2 + daemon._state_lock = threading.Lock() + + bar_time = pd.Timestamp("2026-07-20 10:00:00", tz="UTC") + df = pd.DataFrame( + {"Open": [2400.0], "High": [2405.0], "Low": [2398.0], + "Close": [2403.0], "Volume": [100.0]}, + index=pd.DatetimeIndex([bar_time], name="Datetime")) + + class MockProvider: + def fetch_ohlcv(self, symbol, tf): + return df + + class MockExecutor: + def close_trade(self, trade_id): + raise oanda_executor.OandaAPIError("market halted", 503) + + daemon._provider = MockProvider() + daemon._executor = MockExecutor() + + daemon._state = { + "trades": { + "666": { + "tf": "5min", "direction": "BUY", "entry": 2400.0, + "sl_initial": 2384.0, "current_sl": 2384.0, "risk": 16.0, + "peak": 2400.0, "trail_armed": False, + "trail_arm_r": 0.6, "trail_dist_r": 0.85, + "bars_held": 0, "last_bar_ts": None, "signal_bar_ts": None, + } + }, + "last_entries": {}, + } + + with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: + tmp_path = f.name + orig_state_file = oanda_daemon.STATE_FILE + try: + oanda_daemon.STATE_FILE = tmp_path + daemon._manage_trailing(Timeframe.M5) + finally: + oanda_daemon.STATE_FILE = orig_state_file + os.unlink(tmp_path) + + assert "666" in daemon._state["trades"], \ + "A failed broker close must keep the trade tracked for retry" + + +@test("app_config: invalid oanda section values raise AppConfigError") +def _(): + import tempfile + for bad in ( + {"oanda": {"quality_gate": "Z"}}, + {"oanda": {"lot_size": 0}}, + {"oanda": {"max_hold_bars": -5}}, + {"oanda": {"unknown_key": 1}}, + ): + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False) as f: + json.dump(bad, f) + tmp_path = f.name + try: + app_config.load_app_config(tmp_path) + assert False, f"Should have raised for {bad}" + except app_config.AppConfigError: + pass + finally: + os.unlink(tmp_path) + + +@test("oanda_daemon: boundary timing matches telegram_bot") +def _(): + from telegram_bot import _seconds_until_boundary as tb_boundary + from oanda_daemon import _seconds_until_boundary as od_boundary + for interval in (5, 15, 60, 240): + tb_val = tb_boundary(interval, buffer=10) + od_val = od_boundary(interval, buffer=10) + assert abs(tb_val - od_val) < 0.01, \ + f"Boundary timing mismatch for {interval}min: {tb_val} vs {od_val}" + + +@test("oanda_daemon: state file round-trip") +def _(): + import tempfile + state = { + "trades": { + "999": { + "tf": "M15", + "direction": "BUY", + "entry": 2400.0, + "sl_initial": 2384.0, + "current_sl": 2392.0, + "risk": 16.0, + "peak": 2412.0, + "trail_armed": True, + "trail_arm_r": 0.6, + "trail_dist_r": 0.75, + "bars_held": 25, + } + } + } + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False) as f: + tmp_path = f.name + + orig_state_file = oanda_daemon.STATE_FILE + try: + oanda_daemon.STATE_FILE = tmp_path + oanda_daemon._save_state(state) + loaded = oanda_daemon._load_state() + assert loaded["trades"]["999"]["entry"] == 2400.0 + assert loaded["trades"]["999"]["trail_armed"] is True + assert loaded["trades"]["999"]["bars_held"] == 25 + finally: + oanda_daemon.STATE_FILE = orig_state_file + os.unlink(tmp_path) + + +@test("app_config: oanda in _KNOWN_TOP (no unknown-key error)") +def _(): + assert "oanda" in app_config._KNOWN_TOP, \ + "'oanda' must be in _KNOWN_TOP to avoid config.json validation errors" + + +@test("app_config: oanda resolver helpers return defaults without config") +def _(): + url = app_config.oanda_base_url() + assert "oanda.com" in url, f"Expected Oanda URL, got {url}" + + +@test("oanda_provider: HTTP error surfaces status code and body") +def _(): + mock_resp = type("MockResp", (), { + "status_code": 401, + "text": '{"errorMessage":"Insufficient authorization to perform request"}', + })() + + provider = oanda_provider.OandaDataProvider.__new__( + oanda_provider.OandaDataProvider) + provider._base_url = "https://test" + provider._spread_cache = {} + provider._session = type("S", (), { + "get": lambda self, *a, **k: mock_resp, + "headers": {}, + })() + + try: + provider._fetch_oanda_candles("XAU/USD", Timeframe.H1, 300) + assert False, "Should have raised OandaDataError" + except oanda_provider.OandaDataError as e: + msg = str(e) + assert "401" in msg, f"Status code missing from error: {msg}" + assert "Insufficient authorization" in msg, \ + f"Body detail missing from error: {msg}" + + +@test("oanda_provider: boundary-retry failure keeps original data") +def _(): + provider = oanda_provider.OandaDataProvider.__new__( + oanda_provider.OandaDataProvider) + provider.config = PlatformConfig() + provider._cache = {} + provider._cache_timestamps = {} + provider._cache_lock = threading.Lock() + provider._spread_cache = {} + provider.FRESH_FETCH_RETRY_SECONDS = 0 # instance override — no sleep + + call_count = {"n": 0} + idx = pd.date_range("2026-07-01", periods=250, freq="5min", tz="UTC") + original_df = pd.DataFrame({ + "Open": 2400.0, "High": 2401.0, "Low": 2399.0, + "Close": 2400.5, "Volume": 100.0, + }, index=idx) + + def _fetch_oanda_candles(symbol, timeframe, count): + call_count["n"] += 1 + if call_count["n"] == 1: + return original_df + raise oanda_provider.OandaDataError("boom") + + def _validate_and_clean(df, symbol, timeframe): + return df + + def _has_latest_closed_bar(df, timeframe): + return False # forces the retry path every call + + def _is_data_fresh(df, symbol, timeframe): + return True + + # Plain functions assigned to the instance bypass the descriptor + # protocol — they run unbound, so they take exactly the params + # fetch_ohlcv passes (no self). + provider._fetch_oanda_candles = _fetch_oanda_candles + provider._validate_and_clean = _validate_and_clean + provider._has_latest_closed_bar = _has_latest_closed_bar + provider._is_data_fresh = _is_data_fresh + + result = provider.fetch_ohlcv("XAU/USD", Timeframe.M5) + + assert call_count["n"] == 2, \ + f"Expected original fetch + one retry attempt, got {call_count['n']} calls" + assert len(result) == 250, \ + f"Retry failure must keep the original 250-bar df, got {len(result)}" + + +@test("oanda_executor: non-JSON response raises OandaAPIError with status") +def _(): + def _bad_json(self): + raise ValueError("Expecting value: line 1 column 1 (char 0)") + + mock_resp = type("MockResp", (), { + "status_code": 200, + "text": "gateway error", + "json": _bad_json, + })() + + executor = oanda_executor.OandaExecutor.__new__( + oanda_executor.OandaExecutor) + executor._account_id = "test-account" + executor._base_url = "https://test" + executor._session = type("S", (), { + "request": lambda self, method, url, **kwargs: mock_resp, + "headers": {}, + })() + + try: + executor.get_open_trades() + assert False, "Should have raised OandaAPIError" + except oanda_executor.OandaAPIError as e: + assert e.status_code == 200, f"Expected status 200, got {e.status_code}" + assert "non-JSON" in str(e), f"Expected non-JSON in message: {e}" + + +@test("oanda_daemon: boot fails fast on rejected credentials (4xx)") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + + class MockExecutor401: + def get_account_summary(self): + raise oanda_executor.OandaAPIError("unauthorized", 401) + + daemon._executor = MockExecutor401() + try: + daemon._verify_account_access() + assert False, "A 4xx credential rejection must raise SystemExit" + except SystemExit: + pass + + class MockExecutorConnFail: + def get_account_summary(self): + raise oanda_executor.OandaAPIError("conn refused", 0) + + daemon._executor = MockExecutorConnFail() + daemon._verify_account_access() # transient — must NOT raise + + +@test("oanda_daemon: heartbeat never raises and logs on account failure") +def _(): + daemon = oanda_daemon.TradingDaemon.__new__(oanda_daemon.TradingDaemon) + daemon._state = {"trades": {}, "last_entries": {}} + daemon._state_lock = threading.Lock() + + class MockExecutorDown: + def get_account_summary(self): + raise oanda_executor.OandaAPIError("down", 503) + + daemon._executor = MockExecutorDown() + daemon._heartbeat_once() # must not raise + + class MockExecutorUp: + def get_account_summary(self): + return {"balance": "1000"} + + daemon._executor = MockExecutorUp() + daemon._heartbeat_once() # must not raise + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════