gavindiaz/fx-quant
1
0
Fork 0
mirror of https://github.com/BrentNeale1/fx-quant.git synced 2026-08-01 12:57:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add backtesting engine with SMA cross strategy

Browse Source 
Reads features from Supabase, simulates SMA(3)/SMA(20) crossover signals,
and outputs P&L, drawdown, and trade logs per instrument/granularity.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Brent Neale
2026-02-16 15:58:00 +10:00
parent 31c74d4e5f
commit 187ebe9a9e
5 changed files with 412 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backtester.py
+343
View File
@@ -0,0 +1,343 @@
# src/backtester.py
"""
Backtesting engine for fx-quant.
Reads features from Supabase, simulates the SMA cross strategy,
and produces P&L, drawdown, and trade log outputs.
"""
import os
import math
from pathlib import Path
import pandas as pd
import numpy as np
from supabase import create_client
from config_loader import load_config, get_project_root
# ---------------------------------------------------------------------------
# Data fetching
# ---------------------------------------------------------------------------
def fetch_candles_from_supabase(instrument, granularity, supabase_client, table):
"""
Query Supabase fx_candles table filtered by instrument + granularity.
Returns a pandas DataFrame indexed by time, sorted ascending.
Drops rows where sma columns are NaN (warmup period).
"""
print(f"Fetching {instrument} / {granularity} from Supabase...")
# Supabase JS-style pagination: fetch in pages of 1000
all_rows = []
page_size = 1000
offset = 0
while True:
resp = (
supabase_client.table(table)
.select("*")
.eq("instrument", instrument)
.eq("granularity", granularity)
.order("time", desc=False)
.range(offset, offset + page_size - 1)
.execute()
)
rows = resp.data or []
all_rows.extend(rows)
if len(rows) < page_size:
break
offset += page_size
if not all_rows:
print(f" No data found for {instrument} / {granularity}.")
return pd.DataFrame()
df = pd.DataFrame(all_rows)
df["time"] = pd.to_datetime(df["time"])
df = df.set_index("time").sort_index()
# Convert numeric columns from possible string types
numeric_cols = [c for c in df.columns if c not in ("instrument", "granularity")]
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors="coerce")
print(f" Fetched {len(df)} rows.")
return df
# ---------------------------------------------------------------------------
# Signal generation
# ---------------------------------------------------------------------------
def generate_signals(df, strategy_cfg):
"""
Generate trading signals based on strategy config.
Currently supports 'sma_cross' rule only.
Signal logic (long-only / flat):
sma_short > sma_long → signal = 1 (long)
sma_short < sma_long → signal = 0 (flat)
Position changes only when signal changes.
"""
rule = strategy_cfg["rule"]
if rule != "sma_cross":
raise ValueError(f"Unsupported strategy rule: {rule}")
short_w = strategy_cfg["params"]["short"]
long_w = strategy_cfg["params"]["long"]
sma_short_col = f"sma_{short_w}"
sma_long_col = f"sma_{long_w}"
# Drop rows where indicator columns are NaN (warmup)
required_cols = [sma_short_col, sma_long_col, "close", "ret"]
for col in required_cols:
if col not in df.columns:
raise KeyError(f"Missing required column: {col}")
df = df.dropna(subset=[sma_short_col, sma_long_col, "ret"]).copy()
# Signal: 1 when short SMA above long SMA, else 0
df["signal"] = np.where(df[sma_short_col] > df[sma_long_col], 1, 0)
# Position changes only on crossover (detect changes)
df["position"] = df["signal"]
return df
# ---------------------------------------------------------------------------
# Backtest engine
# ---------------------------------------------------------------------------
def run_backtest(df, strategy_cfg):
"""
Walk through bars, track position, equity, trades.
Returns dict with equity_curve (Series), trades (list of dicts), metrics (dict).
"""
trade_size_pct = strategy_cfg.get("trade_size_pct_of_equity", 0.01)
max_dd_pct = strategy_cfg.get("max_drawdown_pct", 0.05)
starting_equity = 10_000.0
equity = starting_equity
peak_equity = equity
position = 0 # 0 = flat, 1 = long
position_size = 0.0
stopped = False
equity_curve = []
trades = []
times = df.index.tolist()
signals = df["signal"].values
closes = df["close"].values
rets = df["ret"].values
for i in range(len(df)):
bar_time = times[i]
sig = signals[i]
price = closes[i]
bar_ret = rets[i]
# Check max drawdown stop
if stopped:
equity_curve.append(equity)
continue
# P&L from existing position
if position == 1 and i > 0:
pnl = position_size * bar_ret
equity += pnl
# Update peak and drawdown
if equity > peak_equity:
peak_equity = equity
drawdown = (peak_equity - equity) / peak_equity if peak_equity > 0 else 0.0
# Max drawdown breached — stop trading
if drawdown >= max_dd_pct:
if position == 1:
trades.append({
"time": bar_time,
"side": "FLAT",
"price": price,
"position_size": 0.0,
"equity": round(equity, 2),
"drawdown": round(drawdown, 6),
})
stopped = True
position = 0
position_size = 0.0
equity_curve.append(equity)
print(f" Max drawdown {max_dd_pct:.1%} breached at {bar_time}. Stopping.")
continue
# Position change
if sig != position:
if sig == 1 and position == 0:
# Enter long
position_size = equity * trade_size_pct
trades.append({
"time": bar_time,
"side": "BUY",
"price": price,
"position_size": round(position_size, 2),
"equity": round(equity, 2),
"drawdown": round(drawdown, 6),
})
elif sig == 0 and position == 1:
# Exit long
trades.append({
"time": bar_time,
"side": "FLAT",
"price": price,
"position_size": 0.0,
"equity": round(equity, 2),
"drawdown": round(drawdown, 6),
})
position_size = 0.0
position = sig
equity_curve.append(equity)
equity_series = pd.Series(equity_curve, index=df.index, name="equity")
metrics = compute_metrics(equity_series, trades, starting_equity)
return {
"equity_curve": equity_series,
"trades": trades,
"metrics": metrics,
}
# ---------------------------------------------------------------------------
# Metrics
# ---------------------------------------------------------------------------
def compute_metrics(equity_curve, trades, starting_equity=10_000.0):
"""
Compute summary metrics from equity curve and trade list.
"""
final_equity = equity_curve.iloc[-1] if len(equity_curve) > 0 else starting_equity
total_return_pct = ((final_equity - starting_equity) / starting_equity) * 100
# Max drawdown from equity curve
peak = equity_curve.cummax()
dd = (peak - equity_curve) / peak
max_drawdown_pct = dd.max() * 100 if len(dd) > 0 else 0.0
# Trade stats
num_trades = len(trades)
# Count winning round-trips (BUY followed by FLAT with higher equity)
wins = 0
buy_equity = None
for t in trades:
if t["side"] == "BUY":
buy_equity = t["equity"]
elif t["side"] == "FLAT" and buy_equity is not None:
if t["equity"] > buy_equity:
wins += 1
buy_equity = None
round_trips = sum(1 for t in trades if t["side"] == "FLAT")
win_rate = (wins / round_trips * 100) if round_trips > 0 else 0.0
# Sharpe ratio (annualized, from per-bar returns of the equity curve)
eq_returns = equity_curve.pct_change().dropna()
sharpe = 0.0
if len(eq_returns) > 1 and eq_returns.std() > 0:
sharpe = (eq_returns.mean() / eq_returns.std()) * math.sqrt(252 * 24 * 60) # per-minute approx
return {
"total_return_pct": round(total_return_pct, 4),
"max_drawdown_pct": round(max_drawdown_pct, 4),
"num_trades": num_trades,
"round_trips": round_trips,
"win_rate_pct": round(win_rate, 2),
"sharpe_ratio": round(sharpe, 4),
"final_equity": round(final_equity, 2),
}
# ---------------------------------------------------------------------------
# Results output
# ---------------------------------------------------------------------------
def save_results(instrument, granularity, trades, metrics):
"""
Save trade log CSV and print summary metrics.
"""
root = get_project_root()
logs_dir = root / "logs"
logs_dir.mkdir(exist_ok=True)
# Trade log CSV
if trades:
trade_df = pd.DataFrame(trades)
trade_df["instrument"] = instrument
trade_df["granularity"] = granularity
# Reorder columns
cols = ["time", "instrument", "granularity", "side", "price",
"position_size", "equity", "drawdown"]
trade_df = trade_df[cols]
csv_path = logs_dir / f"backtest_trades_{instrument}_{granularity}.csv"
trade_df.to_csv(csv_path, index=False)
print(f" Trade log saved: {csv_path}")
else:
print(" No trades to log.")
# Console summary
print(f"\n --- {instrument} / {granularity} Summary ---")
print(f" Total return: {metrics['total_return_pct']:.4f}%")
print(f" Max drawdown: {metrics['max_drawdown_pct']:.4f}%")
print(f" Trades: {metrics['num_trades']} (round-trips: {metrics['round_trips']})")
print(f" Win rate: {metrics['win_rate_pct']:.2f}%")
print(f" Sharpe ratio: {metrics['sharpe_ratio']:.4f}")
print(f" Final equity: ${metrics['final_equity']:,.2f}")
print()
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
cfg = load_config()
# Supabase client
supabase_url = os.getenv("SUPABASE_URL")
supabase_key = os.getenv("SUPABASE_KEY")
if not supabase_url or not supabase_key:
raise SystemExit("Missing SUPABASE_URL or SUPABASE_KEY in config/.env")
sb = create_client(supabase_url, supabase_key)
table = cfg.get("supabase", {}).get("table", "fx_candles")
strategy_cfg = cfg["strategy"]
instruments = cfg["brokers"][0]["instruments"]
granularities = cfg["data"]["candle_granularities"]
print(f"Backtesting strategy: {strategy_cfg['rule']}")
print(f"Instruments: {instruments}")
print(f"Granularities: {granularities}\n")
for instrument in instruments:
for granularity in granularities:
print(f"=== {instrument} / {granularity} ===")
df = fetch_candles_from_supabase(instrument, granularity, sb, table)
if df.empty:
print(" Skipping — no data.\n")
continue
df = generate_signals(df, strategy_cfg)
if df.empty:
print(" Skipping — no valid rows after warmup.\n")
continue
results = run_backtest(df, strategy_cfg)
save_results(instrument, granularity, results["trades"], results["metrics"])
print("Backtesting complete.")
if __name__ == "__main__":
main()