mirror of
https://github.com/BrentNeale1/fx-quant.git
synced 2026-08-07 07:37:44 +00:00
Add web dashboard, historical data loader, and backtest reporting
- Add Flask dashboard with status, config editor, logs, kill switch, and backtest results pages with monthly P&L breakdowns - Add historical_loader.py for paginated OANDA candle fetching (1yr+) - Update backtester to $100k starting equity, monthly P&L computation, and JSON summary output for dashboard display - Update config to EUR_USD only on M5/M15 with SMA 21/50 strategy - Add backtest.html template with performance metrics and bar charts Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Brent Neale
+66
-6
@@ -6,6 +6,7 @@ and produces P&L, drawdown, and trade log outputs.
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
|
||||
@@ -117,7 +118,7 @@ def run_backtest(df, strategy_cfg):
|
||||
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
|
||||
starting_equity = cfg_equity if (cfg_equity := strategy_cfg.get("starting_equity")) else 100_000.0
|
||||
equity = starting_equity
|
||||
peak_equity = equity
|
||||
position = 0 # 0 = flat, 1 = long
|
||||
@@ -213,7 +214,7 @@ def run_backtest(df, strategy_cfg):
|
||||
# Metrics
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def compute_metrics(equity_curve, trades, starting_equity=10_000.0):
|
||||
def compute_metrics(equity_curve, trades, starting_equity=100_000.0):
|
||||
"""
|
||||
Compute summary metrics from equity curve and trade list.
|
||||
"""
|
||||
@@ -248,6 +249,7 @@ def compute_metrics(equity_curve, trades, starting_equity=10_000.0):
|
||||
sharpe = (eq_returns.mean() / eq_returns.std()) * math.sqrt(252 * 24 * 60) # per-minute approx
|
||||
|
||||
return {
|
||||
"starting_equity": round(starting_equity, 2),
|
||||
"total_return_pct": round(total_return_pct, 4),
|
||||
"max_drawdown_pct": round(max_drawdown_pct, 4),
|
||||
"num_trades": num_trades,
|
||||
@@ -262,20 +264,48 @@ def compute_metrics(equity_curve, trades, starting_equity=10_000.0):
|
||||
# Results output
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def save_results(instrument, granularity, trades, metrics):
|
||||
def compute_monthly_pnl(equity_curve, starting_equity=100_000.0):
|
||||
"""
|
||||
Save trade log CSV and print summary metrics.
|
||||
Compute P&L for each calendar month from the equity curve.
|
||||
Returns list of dicts with month, start_equity, end_equity, pnl, pnl_pct.
|
||||
"""
|
||||
monthly = []
|
||||
# Group by year-month
|
||||
grouped = equity_curve.groupby(equity_curve.index.to_period("M"))
|
||||
prev_end = starting_equity
|
||||
|
||||
for period, group in grouped:
|
||||
end_eq = group.iloc[-1]
|
||||
pnl = end_eq - prev_end
|
||||
pnl_pct = (pnl / prev_end * 100) if prev_end != 0 else 0.0
|
||||
monthly.append({
|
||||
"month": str(period),
|
||||
"start_equity": round(prev_end, 2),
|
||||
"end_equity": round(end_eq, 2),
|
||||
"pnl": round(pnl, 2),
|
||||
"pnl_pct": round(pnl_pct, 4),
|
||||
})
|
||||
prev_end = end_eq
|
||||
|
||||
return monthly
|
||||
|
||||
|
||||
def save_results(instrument, granularity, results, metrics):
|
||||
"""
|
||||
Save trade log CSV, monthly P&L, and a JSON summary for the dashboard.
|
||||
"""
|
||||
root = get_project_root()
|
||||
logs_dir = root / "logs"
|
||||
logs_dir.mkdir(exist_ok=True)
|
||||
|
||||
trades = results["trades"]
|
||||
equity_curve = results["equity_curve"]
|
||||
|
||||
# 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]
|
||||
@@ -285,6 +315,28 @@ def save_results(instrument, granularity, trades, metrics):
|
||||
else:
|
||||
print(" No trades to log.")
|
||||
|
||||
# Monthly P&L
|
||||
starting_equity = metrics.get("starting_equity", 100_000.0)
|
||||
monthly = compute_monthly_pnl(equity_curve, starting_equity)
|
||||
|
||||
# Save JSON summary for dashboard
|
||||
summary = {
|
||||
"instrument": instrument,
|
||||
"granularity": granularity,
|
||||
"run_time": pd.Timestamp.now(tz="UTC").isoformat(),
|
||||
"data_range": {
|
||||
"start": str(equity_curve.index[0]) if len(equity_curve) > 0 else "",
|
||||
"end": str(equity_curve.index[-1]) if len(equity_curve) > 0 else "",
|
||||
"bars": len(equity_curve),
|
||||
},
|
||||
"metrics": metrics,
|
||||
"monthly_pnl": monthly,
|
||||
}
|
||||
json_path = logs_dir / f"backtest_summary_{instrument}_{granularity}.json"
|
||||
with open(json_path, "w") as f:
|
||||
json.dump(summary, f, indent=2, default=str)
|
||||
print(f" Summary saved: {json_path}")
|
||||
|
||||
# Console summary
|
||||
print(f"\n --- {instrument} / {granularity} Summary ---")
|
||||
print(f" Total return: {metrics['total_return_pct']:.4f}%")
|
||||
@@ -293,6 +345,14 @@ def save_results(instrument, granularity, trades, metrics):
|
||||
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}")
|
||||
|
||||
# Monthly P&L table
|
||||
print(f"\n Monthly P&L:")
|
||||
print(f" {'Month':<10} {'Start':>12} {'End':>12} {'P&L':>10} {'%':>8}")
|
||||
print(f" {'-'*54}")
|
||||
for m in monthly:
|
||||
sign = "+" if m["pnl"] >= 0 else ""
|
||||
print(f" {m['month']:<10} ${m['start_equity']:>11,.2f} ${m['end_equity']:>11,.2f} {sign}${m['pnl']:>8,.2f} {sign}{m['pnl_pct']:.2f}%")
|
||||
print()
|
||||
|
||||
|
||||
@@ -334,7 +394,7 @@ def main():
|
||||
continue
|
||||
|
||||
results = run_backtest(df, strategy_cfg)
|
||||
save_results(instrument, granularity, results["trades"], results["metrics"])
|
||||
save_results(instrument, granularity, results, results["metrics"])
|
||||
|
||||
print("Backtesting complete.")
|
||||
|
||||
|
||||
@@ -0,0 +1,304 @@
|
||||
# src/dashboard.py
|
||||
"""
|
||||
Web dashboard for fx-quant trading bot.
|
||||
Provides config editing, status monitoring, log viewing, and kill switch control.
|
||||
"""
|
||||
|
||||
import os
|
||||
import csv
|
||||
import json
|
||||
import shutil
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from functools import wraps
|
||||
|
||||
import yaml
|
||||
from flask import Flask, render_template, request, redirect, url_for, flash, Response
|
||||
|
||||
app = Flask(__name__, template_folder=str(Path(__file__).resolve().parent.parent / "templates"))
|
||||
app.secret_key = os.urandom(24)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Paths
|
||||
# ---------------------------------------------------------------------------
|
||||
APP_ROOT = Path(__file__).resolve().parent.parent
|
||||
CONFIG_PATH = APP_ROOT / "config" / "system.yaml"
|
||||
ENV_PATH = APP_ROOT / "config" / ".env"
|
||||
RELOAD_SIGNAL = APP_ROOT / "RELOAD_CONFIG"
|
||||
KILL_SWITCH_FILE = APP_ROOT / "STOP_ALL_TRADING"
|
||||
ORDER_LOG = APP_ROOT / "logs" / "order_log.csv"
|
||||
AI_LOG = APP_ROOT / "logs" / "ai_decisions.csv"
|
||||
|
||||
# When running in Docker, paths are under /app
|
||||
if Path("/app/config/system.yaml").exists():
|
||||
CONFIG_PATH = Path("/app/config/system.yaml")
|
||||
ENV_PATH = Path("/app/config/.env")
|
||||
RELOAD_SIGNAL = Path("/app/RELOAD_CONFIG")
|
||||
KILL_SWITCH_FILE = Path("/app/STOP_ALL_TRADING")
|
||||
ORDER_LOG = Path("/app/logs/order_log.csv")
|
||||
AI_LOG = Path("/app/logs/ai_decisions.csv")
|
||||
|
||||
# Whitelisted OANDA instruments for validation
|
||||
VALID_INSTRUMENTS = [
|
||||
"EUR_USD", "USD_JPY", "GBP_USD", "USD_CHF", "AUD_USD",
|
||||
"USD_CAD", "NZD_USD", "EUR_GBP", "EUR_JPY", "GBP_JPY",
|
||||
"EUR_CHF", "AUD_JPY", "CHF_JPY", "EUR_AUD", "EUR_CAD",
|
||||
"EUR_NZD", "GBP_AUD", "GBP_CAD", "GBP_CHF", "GBP_NZD",
|
||||
"AUD_CAD", "AUD_CHF", "AUD_NZD", "CAD_CHF", "CAD_JPY",
|
||||
"NZD_CAD", "NZD_CHF", "NZD_JPY",
|
||||
]
|
||||
|
||||
VALID_GRANULARITIES = ["S5", "S10", "S15", "S30", "M1", "M2", "M4", "M5",
|
||||
"M10", "M15", "M30", "H1", "H2", "H3", "H4", "H6",
|
||||
"H8", "H12", "D", "W", "M"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Auth
|
||||
# ---------------------------------------------------------------------------
|
||||
def check_auth(username, password):
|
||||
dashboard_pw = os.environ.get("DASHBOARD_PASSWORD", "changeme")
|
||||
return username == "admin" and password == dashboard_pw
|
||||
|
||||
|
||||
def authenticate():
|
||||
return Response(
|
||||
"Login required.", 401,
|
||||
{"WWW-Authenticate": 'Basic realm="fx-quant dashboard"'},
|
||||
)
|
||||
|
||||
|
||||
def requires_auth(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
auth = request.authorization
|
||||
if not auth or not check_auth(auth.username, auth.password):
|
||||
return authenticate()
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
def load_config():
|
||||
with open(CONFIG_PATH, "r") as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
|
||||
def save_config(cfg):
|
||||
"""Save config with backup."""
|
||||
backup = CONFIG_PATH.with_suffix(".yaml.backup")
|
||||
shutil.copy2(CONFIG_PATH, backup)
|
||||
with open(CONFIG_PATH, "w") as f:
|
||||
yaml.dump(cfg, f, default_flow_style=False, sort_keys=False)
|
||||
# Signal bot to reload
|
||||
RELOAD_SIGNAL.touch()
|
||||
|
||||
|
||||
def read_csv_tail(csv_path, max_rows=50):
|
||||
"""Read last N rows from a CSV file, return (headers, rows)."""
|
||||
if not csv_path.exists():
|
||||
return [], []
|
||||
with open(csv_path, "r") as f:
|
||||
reader = csv.reader(f)
|
||||
rows = list(reader)
|
||||
if not rows:
|
||||
return [], []
|
||||
headers = rows[0]
|
||||
data = rows[1:]
|
||||
return headers, data[-max_rows:]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Validation
|
||||
# ---------------------------------------------------------------------------
|
||||
def validate_config(cfg):
|
||||
"""Validate config values. Returns list of error strings."""
|
||||
errors = []
|
||||
|
||||
# Instruments
|
||||
instruments = cfg.get("brokers", [{}])[0].get("instruments", [])
|
||||
for inst in instruments:
|
||||
if inst not in VALID_INSTRUMENTS:
|
||||
errors.append(f"Invalid instrument: {inst}")
|
||||
|
||||
# Granularities
|
||||
grans = cfg.get("data", {}).get("candle_granularities", [])
|
||||
for g in grans:
|
||||
if g not in VALID_GRANULARITIES:
|
||||
errors.append(f"Invalid granularity: {g}")
|
||||
|
||||
# Numeric ranges
|
||||
strategy = cfg.get("strategy", {})
|
||||
params = strategy.get("params", {})
|
||||
if params.get("short", 1) < 1:
|
||||
errors.append("SMA short period must be >= 1")
|
||||
if params.get("long", 2) < 2:
|
||||
errors.append("SMA long period must be >= 2")
|
||||
if params.get("short", 1) >= params.get("long", 2):
|
||||
errors.append("SMA short period must be less than long period")
|
||||
|
||||
trade_size = strategy.get("trade_size_pct_of_equity", 0.01)
|
||||
if not (0.001 <= trade_size <= 0.1):
|
||||
errors.append("trade_size_pct_of_equity must be between 0.001 and 0.1")
|
||||
|
||||
execution = cfg.get("execution", {})
|
||||
if execution.get("max_positions", 1) < 1:
|
||||
errors.append("max_positions must be >= 1")
|
||||
if execution.get("interval_seconds", 10) < 10:
|
||||
errors.append("interval_seconds must be >= 10")
|
||||
|
||||
ai = cfg.get("ai", {})
|
||||
threshold = ai.get("confidence_threshold", 0.5)
|
||||
if not (0.0 <= threshold <= 1.0):
|
||||
errors.append("confidence_threshold must be between 0.0 and 1.0")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Routes
|
||||
# ---------------------------------------------------------------------------
|
||||
@app.route("/")
|
||||
@requires_auth
|
||||
def index():
|
||||
cfg = load_config()
|
||||
kill_active = KILL_SWITCH_FILE.exists()
|
||||
|
||||
# Read recent orders for activity feed
|
||||
_, recent_orders = read_csv_tail(ORDER_LOG, max_rows=10)
|
||||
|
||||
return render_template("index.html",
|
||||
cfg=cfg,
|
||||
kill_active=kill_active,
|
||||
recent_orders=recent_orders)
|
||||
|
||||
|
||||
@app.route("/config", methods=["GET", "POST"])
|
||||
@requires_auth
|
||||
def config_editor():
|
||||
cfg = load_config()
|
||||
|
||||
if request.method == "POST":
|
||||
# Parse form into config structure
|
||||
try:
|
||||
# Instruments
|
||||
instruments_raw = request.form.get("instruments", "").strip()
|
||||
instruments = [i.strip() for i in instruments_raw.split(",") if i.strip()]
|
||||
cfg["brokers"][0]["instruments"] = instruments
|
||||
|
||||
# Granularities
|
||||
grans_raw = request.form.get("granularities", "").strip()
|
||||
grans = [g.strip() for g in grans_raw.split(",") if g.strip()]
|
||||
cfg["data"]["candle_granularities"] = grans
|
||||
|
||||
# Candle count
|
||||
cfg["data"]["candle_count"] = int(request.form.get("candle_count", 200))
|
||||
|
||||
# Features
|
||||
cfg["features"]["sma_windows"] = _parse_int_list(request.form.get("sma_windows", "3,20"))
|
||||
cfg["features"]["ema_windows"] = _parse_int_list(request.form.get("ema_windows", "20"))
|
||||
cfg["features"]["rsi_period"] = int(request.form.get("rsi_period", 14))
|
||||
cfg["features"]["atr_period"] = int(request.form.get("atr_period", 14))
|
||||
cfg["features"]["vwap_window"] = int(request.form.get("vwap_window", 20))
|
||||
cfg["features"]["volatility_window"] = int(request.form.get("volatility_window", 20))
|
||||
|
||||
# Strategy
|
||||
cfg["strategy"]["rule"] = request.form.get("strategy_rule", "sma_cross")
|
||||
cfg["strategy"]["params"]["short"] = int(request.form.get("sma_short", 3))
|
||||
cfg["strategy"]["params"]["long"] = int(request.form.get("sma_long", 20))
|
||||
cfg["strategy"]["trade_size_pct_of_equity"] = float(request.form.get("trade_size_pct", 0.01))
|
||||
cfg["strategy"]["max_drawdown_pct"] = float(request.form.get("max_drawdown_pct", 0.05))
|
||||
|
||||
# AI
|
||||
cfg["ai"]["confidence_threshold"] = float(request.form.get("confidence_threshold", 0.85))
|
||||
cfg["ai"]["ensemble_models"] = [m.strip() for m in request.form.get("ensemble_models", "").split(",") if m.strip()]
|
||||
cfg["ai"]["sanity_checks"]["rsi_overbought"] = int(request.form.get("rsi_overbought", 80))
|
||||
cfg["ai"]["sanity_checks"]["rsi_oversold"] = int(request.form.get("rsi_oversold", 20))
|
||||
cfg["ai"]["sanity_checks"]["volatility_multiplier"] = float(request.form.get("volatility_multiplier", 3.0))
|
||||
|
||||
# Execution
|
||||
cfg["execution"]["paper_mode"] = request.form.get("paper_mode") == "on"
|
||||
cfg["execution"]["canary_size_pct"] = float(request.form.get("canary_size_pct", 0.01))
|
||||
cfg["execution"]["max_positions"] = int(request.form.get("max_positions", 3))
|
||||
cfg["execution"]["interval_seconds"] = int(request.form.get("interval_seconds", 60))
|
||||
|
||||
# Validate
|
||||
errors = validate_config(cfg)
|
||||
if errors:
|
||||
for e in errors:
|
||||
flash(e, "danger")
|
||||
return render_template("config.html", cfg=cfg,
|
||||
valid_instruments=VALID_INSTRUMENTS,
|
||||
valid_granularities=VALID_GRANULARITIES)
|
||||
|
||||
save_config(cfg)
|
||||
flash("Config saved. Bot will reload on next loop iteration.", "success")
|
||||
return redirect(url_for("config_editor"))
|
||||
|
||||
except (ValueError, KeyError) as e:
|
||||
flash(f"Invalid input: {e}", "danger")
|
||||
return render_template("config.html", cfg=cfg,
|
||||
valid_instruments=VALID_INSTRUMENTS,
|
||||
valid_granularities=VALID_GRANULARITIES)
|
||||
|
||||
return render_template("config.html", cfg=cfg,
|
||||
valid_instruments=VALID_INSTRUMENTS,
|
||||
valid_granularities=VALID_GRANULARITIES)
|
||||
|
||||
|
||||
@app.route("/logs")
|
||||
@requires_auth
|
||||
def logs():
|
||||
order_headers, order_rows = read_csv_tail(ORDER_LOG, max_rows=100)
|
||||
ai_headers, ai_rows = read_csv_tail(AI_LOG, max_rows=100)
|
||||
# Reverse so newest first
|
||||
order_rows = list(reversed(order_rows))
|
||||
ai_rows = list(reversed(ai_rows))
|
||||
return render_template("logs.html",
|
||||
order_headers=order_headers,
|
||||
order_rows=order_rows,
|
||||
ai_headers=ai_headers,
|
||||
ai_rows=ai_rows)
|
||||
|
||||
|
||||
@app.route("/backtest")
|
||||
@requires_auth
|
||||
def backtest():
|
||||
"""Show backtest results from saved JSON summaries."""
|
||||
logs_dir = APP_ROOT / "logs"
|
||||
summaries = []
|
||||
for f in sorted(logs_dir.glob("backtest_summary_*.json")):
|
||||
with open(f, "r") as fh:
|
||||
summaries.append(json.load(fh))
|
||||
return render_template("backtest.html", summaries=summaries)
|
||||
|
||||
|
||||
@app.route("/killswitch", methods=["POST"])
|
||||
@requires_auth
|
||||
def killswitch():
|
||||
action = request.form.get("action")
|
||||
if action == "activate":
|
||||
KILL_SWITCH_FILE.touch()
|
||||
flash("Kill switch ACTIVATED. All trading halted.", "warning")
|
||||
elif action == "deactivate":
|
||||
if KILL_SWITCH_FILE.exists():
|
||||
KILL_SWITCH_FILE.unlink()
|
||||
flash("Kill switch deactivated. Trading will resume on next loop.", "success")
|
||||
return redirect(url_for("index"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
def _parse_int_list(s):
|
||||
"""Parse comma-separated string into list of ints."""
|
||||
return [int(x.strip()) for x in s.split(",") if x.strip()]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
if __name__ == "__main__":
|
||||
app.run(host="0.0.0.0", port=5000, debug=False)
|
||||
@@ -0,0 +1,129 @@
|
||||
# src/historical_loader.py
|
||||
"""
|
||||
Fetch ~1 year of historical OANDA candles in paginated chunks (max 5000 per request),
|
||||
compute features, and upload to Supabase.
|
||||
|
||||
Usage:
|
||||
python src/historical_loader.py
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
import requests
|
||||
import pandas as pd
|
||||
|
||||
from config_loader import load_config
|
||||
from data_engine import candles_to_df, build_all_features
|
||||
from supabase_upload import upload_dataframe
|
||||
|
||||
cfg = load_config()
|
||||
|
||||
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"
|
||||
)
|
||||
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
|
||||
|
||||
|
||||
def fetch_candles_chunk(instrument, granularity, from_dt, to_dt, count=5000):
|
||||
"""Fetch up to `count` candles between from_dt and to_dt."""
|
||||
params = {
|
||||
"granularity": granularity,
|
||||
"from": from_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||||
"to": to_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||||
}
|
||||
url = f"{BASE}/v3/instruments/{instrument}/candles"
|
||||
r = requests.get(url, headers=HEADERS, params=params)
|
||||
r.raise_for_status()
|
||||
return r.json().get("candles", [])
|
||||
|
||||
|
||||
def fetch_all_candles(instrument, granularity, days_back=365):
|
||||
"""
|
||||
Paginate through OANDA history in chunks of 5000 candles.
|
||||
Returns a single merged DataFrame with all candles.
|
||||
"""
|
||||
end = datetime.now(timezone.utc)
|
||||
start = end - timedelta(days=days_back)
|
||||
|
||||
# Determine candle duration for stepping forward
|
||||
gran_minutes = {
|
||||
"M1": 1, "M5": 5, "M15": 15, "M30": 30,
|
||||
"H1": 60, "H4": 240, "D": 1440, "W": 10080,
|
||||
}
|
||||
minutes = gran_minutes.get(granularity, 5)
|
||||
chunk_duration = timedelta(minutes=minutes * 4999) # just under 5000 candles
|
||||
|
||||
all_candles = []
|
||||
cursor = start
|
||||
chunk_num = 0
|
||||
|
||||
while cursor < end:
|
||||
chunk_end = min(cursor + chunk_duration, end)
|
||||
chunk_num += 1
|
||||
print(f" Chunk {chunk_num}: {cursor.strftime('%Y-%m-%d %H:%M')} -> {chunk_end.strftime('%Y-%m-%d %H:%M')} ...", end=" ")
|
||||
|
||||
candles = fetch_candles_chunk(instrument, granularity, cursor, chunk_end)
|
||||
print(f"{len(candles)} candles")
|
||||
|
||||
if candles:
|
||||
all_candles.extend(candles)
|
||||
# Move cursor past the last candle we received
|
||||
last_time = pd.to_datetime(candles[-1]["time"])
|
||||
cursor = last_time.to_pydatetime().replace(tzinfo=timezone.utc) + timedelta(minutes=minutes)
|
||||
else:
|
||||
# No data in this window, skip forward
|
||||
cursor = chunk_end
|
||||
|
||||
# Be polite to the API
|
||||
time.sleep(0.5)
|
||||
|
||||
print(f" Total raw candles fetched: {len(all_candles)}")
|
||||
|
||||
if not all_candles:
|
||||
return pd.DataFrame()
|
||||
|
||||
# Deduplicate by time (overlapping chunks)
|
||||
df = candles_to_df(all_candles)
|
||||
df = df[~df.index.duplicated(keep="last")]
|
||||
df = df.sort_index()
|
||||
print(f" After dedup: {len(df)} candles")
|
||||
print(f" Range: {df.index[0]} -> {df.index[-1]}")
|
||||
|
||||
return df
|
||||
|
||||
|
||||
def main():
|
||||
feature_cfg = cfg.get("features", {})
|
||||
broker = cfg["brokers"][0]
|
||||
instruments = broker["instruments"]
|
||||
granularities = cfg["data"]["candle_granularities"]
|
||||
|
||||
for instrument in instruments:
|
||||
for granularity in granularities:
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Fetching {instrument} / {granularity} — last 365 days")
|
||||
print(f"{'='*60}")
|
||||
|
||||
df = fetch_all_candles(instrument, granularity, days_back=365)
|
||||
if df.empty:
|
||||
print(f" No data. Skipping.")
|
||||
continue
|
||||
|
||||
print(f" Computing features...")
|
||||
df = build_all_features(df, config=feature_cfg)
|
||||
|
||||
print(f" Uploading to Supabase...")
|
||||
upload_dataframe(df, instrument=instrument, granularity=granularity, chunk_size=500)
|
||||
print(f" Done: {instrument} / {granularity}")
|
||||
|
||||
print(f"\nAll historical data loaded.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -465,6 +465,18 @@ if __name__ == "__main__":
|
||||
print(f"Running on {interval}s loop. Press Ctrl+C to stop.\n")
|
||||
while True:
|
||||
try:
|
||||
# Check for config reload signal from dashboard
|
||||
reload_signal = get_project_root() / "RELOAD_CONFIG"
|
||||
if reload_signal.exists():
|
||||
print("\n*** CONFIG RELOAD REQUESTED ***")
|
||||
try:
|
||||
reload_signal.unlink()
|
||||
except OSError:
|
||||
pass
|
||||
cfg = load_config()
|
||||
interval = cfg.get("execution", {}).get("interval_seconds", 60)
|
||||
print(f"Config reloaded. Interval now {interval}s.\n")
|
||||
|
||||
main()
|
||||
print(f"\nSleeping {interval}s until next run...\n")
|
||||
time.sleep(interval)
|
||||
|
||||
Reference in New Issue
Block a user