mirror of
https://github.com/BrentNeale1/fx-quant.git
synced 2026-08-17 12:08:07 +00:00
Add config-driven system.yaml and multi-window indicators
- 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 <noreply@anthropic.com>
This commit is contained in:
Brent Neale
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
256dffaae1
commit
0d0c363bfa
@@ -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
|
||||
<project_root>/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
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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.")
|
||||
Reference in New Issue
Block a user