Harwinder Sekhon
d669227ddc
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
362 lines
13 KiB
Python
362 lines
13 KiB
Python
"""
|
|
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
|