From 0d0c363bfac443271e37571885920357a22633c6 Mon Sep 17 00:00:00 2001 From: Brent Neale Date: Mon, 16 Feb 2026 15:22:07 +1000 Subject: [PATCH] Add config-driven system.yaml and multi-window indicators MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Create config/system.yaml with full master prompt template - Create src/config_loader.py to load YAML config and .env - Update data_engine.build_all_features() to support lists of windows (e.g. sma_windows: [3, 20] produces sma_3 and sma_20 columns) - Rewrite get_candles.py to loop over all instruments × granularities from config instead of hardcoding EUR_USD/M5 - Fix supabase_upload.py indentation bug and wire up config for table name Co-Authored-By: Claude Opus 4.6 --- config/system.yaml | 50 ++++++++++++ src/config_loader.py | 44 +++++++++++ src/data_engine.py | 127 ++++++++++++++++++++++++++++++ src/get_candles.py | 56 ++++++++++++++ src/supabase_upload.py | 170 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 447 insertions(+) create mode 100644 config/system.yaml create mode 100644 src/config_loader.py create mode 100644 src/data_engine.py create mode 100644 src/get_candles.py create mode 100644 src/supabase_upload.py diff --git a/config/system.yaml b/config/system.yaml new file mode 100644 index 0000000..5c076f4 --- /dev/null +++ b/config/system.yaml @@ -0,0 +1,50 @@ +general: + project_name: "fx-quant" + timezone: "UTC" + +brokers: + - name: "oanda" + type: "oanda_v20" + enabled: true + instruments: + - "EUR_USD" + - "USD_JPY" + +data: + candle_count: 200 + candle_granularities: + - "M1" + - "M5" + tick_export: false # true to capture raw ticks + +features: + sma_windows: [3, 20] + ema_windows: [20] + rsi_period: 14 + atr_period: 14 + vwap_window: 20 + volatility_window: 20 + +ai: + model: "local-ensemble" # example: 'openai', 'local-llm', 'ensemble' + confidence_threshold: 0.85 + retriever_enabled: true + retriever_source: "supabase" + +strategy: + rule: "sma_cross" # human-readable name of the rule + params: + short: 3 + long: 20 + trade_size_pct_of_equity: 0.01 + max_drawdown_pct: 0.05 + +execution: + paper_mode: true + canary_size_pct: 0.01 + max_positions: 3 + +supabase: + url: "https://.supabase.co" + key_env_name: "SUPABASE_KEY" # key stored in .env + table: "fx_candles" diff --git a/src/config_loader.py b/src/config_loader.py new file mode 100644 index 0000000..c2e562c --- /dev/null +++ b/src/config_loader.py @@ -0,0 +1,44 @@ +# src/config_loader.py +""" +Loads config/system.yaml and config/.env for the fx-quant project. +""" + +from pathlib import Path +from dotenv import load_dotenv +import yaml + + +def get_project_root() -> Path: + """Return the fx-quant project root (parent of src/).""" + return Path(__file__).resolve().parents[1] + + +def load_config(path=None) -> dict: + """ + Load config/system.yaml and config/.env. + + Parameters + ---------- + path : str or Path, optional + Explicit path to a YAML config file. Defaults to + /config/system.yaml. + + Returns + ------- + dict + Parsed YAML contents. + """ + root = get_project_root() + + # Always load .env as a side effect + env_path = root / "config" / ".env" + if env_path.exists(): + load_dotenv(env_path) + + if path is None: + path = root / "config" / "system.yaml" + + with open(path, "r") as f: + cfg = yaml.safe_load(f) + + return cfg diff --git a/src/data_engine.py b/src/data_engine.py new file mode 100644 index 0000000..863aab8 --- /dev/null +++ b/src/data_engine.py @@ -0,0 +1,127 @@ +# src/data_engine.py +""" +Feature builder for FX candles. +Inputs: list of OANDA candle dicts (with 'mid' prices and 'volume'). +Outputs: pandas DataFrame indexed by time with indicator columns. +""" + +import pandas as pd +import numpy as np + + +def candles_to_df(candles): + rows = [] + for c in candles: + mid = c["mid"] + rows.append({ + "time": c["time"], + "open": float(mid["o"]), + "high": float(mid["h"]), + "low": float(mid["l"]), + "close": float(mid["c"]), + "volume": c.get("volume", 0) + }) + + df = pd.DataFrame(rows) + df["time"] = pd.to_datetime(df["time"]) + df = df.set_index("time").sort_index() + return df + + +def add_sma(df, period=20, col="close"): + df[f"sma_{period}"] = df[col].rolling(period).mean() + return df + + +def add_ema(df, period=20, col="close"): + df[f"ema_{period}"] = df[col].ewm(span=period, adjust=False).mean() + return df + + +def add_returns(df, col="close"): + # simple pct returns and log returns + df["ret"] = df[col].pct_change() + df["logret"] = np.log(df[col] / df[col].shift(1)) + return df + + +def add_rolling_volatility(df, period=20, source="logret"): + # annualization is not necessary here; this gives rolling stdev of returns + df[f"vol_{period}"] = df[source].rolling(period).std() + return df + + +def add_rsi(df, period=14, col="close"): + # Classic RSI (Wilder's) + delta = df[col].diff() + up = delta.clip(lower=0) + down = -1 * delta.clip(upper=0) + ma_up = up.rolling(period).mean() + ma_down = down.rolling(period).mean() + rs = ma_up / (ma_down.replace(0, np.nan)) + df[f"rsi_{period}"] = 100 - (100 / (1 + rs)) + return df + + +def add_atr(df, period=14): + # Average True Range + high_low = df["high"] - df["low"] + high_close = (df["high"] - df["close"].shift(1)).abs() + low_close = (df["low"] - df["close"].shift(1)).abs() + tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) + df["tr"] = tr + df[f"atr_{period}"] = tr.rolling(period).mean() + return df + + +def add_vwap(df, period=20): + """ + Rolling VWAP-like: uses typical price * volume over rolling window. + VWAP usually applies intraday at tick-level; for candle-level a rolling VWAP is a useful proxy. + """ + tp = (df["high"] + df["low"] + df["close"]) / 3.0 + pv = tp * df["volume"] + df["typical_price"] = tp + df["pv"] = pv + # rolling VWAP: sum(price*vol)/sum(vol) + df[f"vwap_{period}"] = df["pv"].rolling(period).sum() / df["volume"].rolling(period).sum() + # protect inf/nan where vol sum == 0 + df[f"vwap_{period}"] = df[f"vwap_{period}"].replace([np.inf, -np.inf], np.nan) + return df + + +def build_all_features(df, config=None): + """ + Run a standard set of features. config is optional dict matching the + 'features' section of system.yaml. Supports lists of windows for + SMA and EMA so multiple columns are produced (e.g. sma_3, sma_20). + + Returns DataFrame with new columns. + """ + if config is None: + config = {} + + # Multi-window indicators (lists) + sma_windows = config.get("sma_windows", [3, 20]) + ema_windows = config.get("ema_windows", [20]) + + # Single-value indicators (scalars) + rsi_p = config.get("rsi_period", 14) + atr_p = config.get("atr_period", 14) + vwap_w = config.get("vwap_window", 20) + vol_w = config.get("volatility_window", 20) + + df = add_returns(df) + + for w in sma_windows: + df = add_sma(df, period=w) + + for w in ema_windows: + df = add_ema(df, period=w) + + df = add_rsi(df, period=rsi_p) + df = add_rolling_volatility(df, period=vol_w, source="logret") + df = add_atr(df, period=atr_p) + df = add_vwap(df, period=vwap_w) + + return df diff --git a/src/get_candles.py b/src/get_candles.py new file mode 100644 index 0000000..d57ed0a --- /dev/null +++ b/src/get_candles.py @@ -0,0 +1,56 @@ +# src/get_candles.py +""" +Fetch OANDA candles for every instrument × granularity defined in +config/system.yaml, build indicator features, and print the result. +""" + +import os +import requests +from config_loader import load_config +from data_engine import candles_to_df, build_all_features + + +def fetch_candles(instrument, granularity, count, base_url, api_key): + """Pull raw candle dicts from the OANDA v20 REST API.""" + url = ( + f"{base_url}/v3/instruments/{instrument}/candles" + f"?count={count}&granularity={granularity}" + ) + headers = {"Authorization": f"Bearer {api_key}"} + r = requests.get(url, headers=headers) + r.raise_for_status() + return r.json()["candles"] + + +def main(): + cfg = load_config() + + # OANDA credentials (loaded into env by config_loader) + api_key = os.getenv("OANDA_API_KEY") + env = os.getenv("OANDA_ENV", "practice") + base_url = ( + "https://api-fxpractice.oanda.com" + if env == "practice" + else "https://api-fxtrade.oanda.com" + ) + + # Config-driven parameters + broker = cfg["brokers"][0] + instruments = broker["instruments"] + granularities = cfg["data"]["candle_granularities"] + count = cfg["data"]["candle_count"] + feature_cfg = cfg.get("features", {}) + + for instrument in instruments: + for granularity in granularities: + print(f"\n--- {instrument} | {granularity} ---") + candles = fetch_candles( + instrument, granularity, count, base_url, api_key + ) + df = candles_to_df(candles) + df = build_all_features(df, config=feature_cfg) + print(df.tail(10).to_string()) + + +if __name__ == "__main__": + main() diff --git a/src/supabase_upload.py b/src/supabase_upload.py new file mode 100644 index 0000000..d32b66b --- /dev/null +++ b/src/supabase_upload.py @@ -0,0 +1,170 @@ +# src/supabase_upload.py +import os +import json +import math +from pathlib import Path +from dotenv import load_dotenv +from supabase import create_client +import pandas as pd +import numpy as np + +from config_loader import load_config + +# Load config (also loads .env as side effect) +cfg = load_config() + +SUPABASE_URL = os.getenv("SUPABASE_URL") +SUPABASE_KEY = os.getenv("SUPABASE_KEY") # use service_role key for server writes +TABLE = cfg.get("supabase", {}).get("table", "fx_candles") + +if not SUPABASE_URL or not SUPABASE_KEY: + raise SystemExit("Missing SUPABASE_URL or SUPABASE_KEY in config/.env") + +supabase = create_client(SUPABASE_URL, SUPABASE_KEY) + + +def df_to_records(df: pd.DataFrame, instrument: str): + """ + Convert DataFrame to a list of dicts suitable for Supabase/Postgres. + Ensures time is ISO string and numeric types are regular Python types. + Replaces inf/-inf with None and converts NaN -> None. + """ + df2 = df.copy() + + # Ensure index is time and reset to a column + if df2.index.name is None: + df2.index.name = "time" + df2 = df2.reset_index() + + # Add instrument column + df2["instrument"] = instrument + + # Convert Timestamp -> ISO string (timezone-aware preserved) + df2["time"] = df2["time"].apply(lambda t: pd.to_datetime(t).isoformat()) + + # Replace infinite values with NaN so they become None later + df2 = df2.replace([np.inf, -np.inf], np.nan) + + def _convert_cell(v): + # None / pandas NA + if pd.isna(v): + return None + + # Native boolean - keep as bool + if isinstance(v, (bool,)): + return bool(v) + + # Integers (numpy or native) + if isinstance(v, (np.integer, int)): + return int(v) + + # Floats (numpy or native) - ensure finite + if isinstance(v, (np.floating, float)): + try: + fv = float(v) + except Exception: + return None + return fv if math.isfinite(fv) else None + + # Strings - keep, but try numeric coercion if they look numeric + if isinstance(v, str): + s = v.strip() + if s.replace('.', '', 1).replace('-', '', 1).isdigit(): + try: + fv = float(s) + return fv if math.isfinite(fv) else None + except Exception: + return s + return s + + # For any other type attempt a last-resort numeric coercion + try: + fv = float(v) + return fv if math.isfinite(fv) else None + except Exception: + try: + return str(v) + except Exception: + return None + + # Apply conversion column-by-column (converts numpy dtypes -> python scalars) + for col in df2.columns: + df2[col] = df2[col].apply(_convert_cell) + + records = df2.to_dict(orient="records") + return records + + +def upload_dataframe(df: pd.DataFrame, instrument="EUR_USD", chunk_size=500): + records = df_to_records(df, instrument) + + # debug: check first record is JSON serializable + if records: + try: + json.dumps(records[0]) + print("First record JSON dump OK.") + except Exception as e: + print("JSON dump failed for first record:", e) + for k, v in records[0].items(): + try: + json.dumps({k: v}) + except Exception as e2: + print( + " BAD KEY:", k, + "VALUE TYPE:", type(v), + "VALUE (truncated):", repr(str(v))[:200], + "->", e2, + ) + + total = len(records) + if total == 0: + print("No rows to upload.") + return + + chunks = math.ceil(total / chunk_size) + print(f"Uploading {total} rows to Supabase in {chunks} chunk(s)...") + + for i in range(chunks): + start = i * chunk_size + end = start + chunk_size + batch = records[start:end] + + resp = supabase.table(TABLE).upsert(batch).execute() + if resp and getattr(resp, "status_code", None) not in (200, 201): + print("Upload chunk error:", resp) + raise SystemExit("Upsert failed") + print(f"Chunk {i+1}/{chunks} uploaded ({len(batch)} rows).") + + +if __name__ == "__main__": + from data_engine import candles_to_df, build_all_features + import requests + + API_KEY = os.getenv("OANDA_API_KEY") + ENV = os.getenv("OANDA_ENV", "practice") + base = ( + "https://api-fxpractice.oanda.com" + if ENV == "practice" + else "https://api-fxtrade.oanda.com" + ) + + feature_cfg = cfg.get("features", {}) + broker = cfg["brokers"][0] + instruments = broker["instruments"] + granularities = cfg["data"]["candle_granularities"] + count = cfg["data"]["candle_count"] + + for instrument in instruments: + for granularity in granularities: + url = f"{base}/v3/instruments/{instrument}/candles?count={count}&granularity={granularity}" + headers = {"Authorization": f"Bearer {API_KEY}"} + r = requests.get(url, headers=headers) + r.raise_for_status() + data = r.json()["candles"] + + df = candles_to_df(data) + df = build_all_features(df, config=feature_cfg) + + upload_dataframe(df, instrument=instrument, chunk_size=200) + + print("Upload completed.")