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. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VUnCqjWBuvKRSjLwHeB9BJ
This commit is contained in:
Harwinder Sekhon
+775
@@ -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()
|
||||
Reference in New Issue
Block a user