diff --git a/QuantTrader/__init__.py b/QuantTrader/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/QuantTrader/__init__.py @@ -0,0 +1 @@ + diff --git a/QuantTrader/__pycache__/__init__.cpython-312.pyc b/QuantTrader/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..23b1467 Binary files /dev/null and b/QuantTrader/__pycache__/__init__.cpython-312.pyc differ diff --git a/QuantTrader/artifacts/README.md b/QuantTrader/artifacts/README.md new file mode 100644 index 0000000..5475acf --- /dev/null +++ b/QuantTrader/artifacts/README.md @@ -0,0 +1,17 @@ +# QuantTrader Artifacts + +This directory stores the research outputs that the trading runtime consumes. + +## Structure +- `config/`: strategy configuration snapshots exported from QuantResearch (YAML). +- `params/`: optimized parameter JSONs (`best_params_*.json`). + +## Manual sync process +1. In `QuantResearch/`, run optimization/backtest scripts to produce updated configs/params. +2. Copy the vetted files into this directory: + - `cp QuantResearch/config/.yaml QuantTrader/artifacts/config/` + - `cp QuantResearch/data/params/best_*.json QuantTrader/artifacts/params/` +3. Commit the new artifacts (or upload to storage) alongside the trading release. +4. Runner processes load configs from `artifacts/config/` and parameters from `artifacts/params/` to ensure live trading uses the approved research snapshot. + +Automating this sync (e.g., via CI) is recommended once the promotion flow stabilizes. diff --git a/QuantTrader/artifacts/config/eurusd_regime.yaml b/QuantTrader/artifacts/config/eurusd_regime.yaml new file mode 100644 index 0000000..03a87dc --- /dev/null +++ b/QuantTrader/artifacts/config/eurusd_regime.yaml @@ -0,0 +1,42 @@ +symbol: EURUSD +csv: data/raw/EURUSD_H1.csv +cash: 100000.0 +qty: 10000 +account_ccy: USD +fast: 20 +slow: 120 +spread: 1.0 +slip: 0.2 +comm: 2.0 +atr_sl: 1.5 +atr_tp: 3.0 +atr_window: 21 +cooldown: 12 +allow_short: true +long_only_above_slow: false +short_only_below_slow: false +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +regime_ema_window: 200 +regime_slope_min: 0.0002 +regime_atr_min: 0.0010 +strategies: + - name: regime_sma + params: + trend_params: + fast_win: 20 + slow_win: 120 + long_only_above_slow: false + slope_lookback: 5 + cooldown: 12 + atr_sl: 1.5 + atr_tp: 3.0 + atr_window: 21 + allow_short: true + short_only_below_slow: false + rsi_period: 14 + rsi_long_thresh: 55 + rsi_short_thresh: 45 + range_mode: mean_revert + range_rsi_high: 65 + range_rsi_low: 35 diff --git a/QuantTrader/artifacts/config/optimized_eurusd_v2_with_rsi.yaml b/QuantTrader/artifacts/config/optimized_eurusd_v2_with_rsi.yaml new file mode 100644 index 0000000..499540d --- /dev/null +++ b/QuantTrader/artifacts/config/optimized_eurusd_v2_with_rsi.yaml @@ -0,0 +1,27 @@ +symbol: EURUSD +csv: data/raw/EURUSD_H1.csv +cash: 100000.0 +qty: 10000 +account_ccy: USD +fast: 20 +slow: 150 +spread: 1.0 +slip: 0.2 +comm: 2.0 +atr_sl: 2.0 +atr_tp: 4.0 +atr_window: 21 +slope_lookback: 5 +cooldown: 24 +allow_short: true +long_only_above_slow: false +short_only_below_slow: false +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +# RSI and trailing +rsi_period: 14 +rsi_long_thresh: 55 +rsi_short_thresh: 45 +enable_trailing: true +trailing_enable_atr_mult: 1.0 +trailing_atr_mult: 0.5 diff --git a/QuantTrader/artifacts/config/optimized_eurusd_v2_with_rsi_best.yaml b/QuantTrader/artifacts/config/optimized_eurusd_v2_with_rsi_best.yaml new file mode 100644 index 0000000..df06073 --- /dev/null +++ b/QuantTrader/artifacts/config/optimized_eurusd_v2_with_rsi_best.yaml @@ -0,0 +1,27 @@ +symbol: EURUSD +csv: data/raw/EURUSD_H1.csv +cash: 100000.0 +qty: 10000 +account_ccy: USD +fast: 20 +slow: 150 +spread: 1.0 +slip: 0.2 +comm: 2.0 +atr_sl: 2.0 +atr_tp: 4.0 +atr_window: 21 +slope_lookback: 5 +cooldown: 24 +allow_short: true +long_only_above_slow: false +short_only_below_slow: false +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +# RSI and trailing - optimized parameters +rsi_period: 14 +rsi_long_thresh: 60 +rsi_short_thresh: 30 +enable_trailing: true +trailing_enable_atr_mult: 0.5 +trailing_atr_mult: 0.3 \ No newline at end of file diff --git a/QuantTrader/artifacts/config/usdjpy_combo.yaml b/QuantTrader/artifacts/config/usdjpy_combo.yaml new file mode 100644 index 0000000..3f2851a --- /dev/null +++ b/QuantTrader/artifacts/config/usdjpy_combo.yaml @@ -0,0 +1,77 @@ +symbol: USDJPY +csv: data/raw/USDJPY_H1.csv +cash: 100000.0 +qty: 10000 +account_ccy: USD +fast: 20 +slow: 120 +spread: 1.2 +slip: 0.2 +comm: 2.0 +atr_sl: 1.2 +atr_tp: 3.5 +atr_window: 21 +cooldown: 24 +allow_short: true +long_only_above_slow: false +short_only_below_slow: false +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +regime_ema_window: 200 +regime_slope_min: 0.0002 +regime_atr_min: 0.0014 +regime_atr_percentile_min: 0.4 +regime_atr_percentile_window: 400 +regime_trend_min_bars: 2 +htf_factor: 4 +htf_ema_window: 60 +htf_rsi_period: 14 +strategies: + - name: regime_sma + params: + trend_params: + fast_win: 20 + slow_win: 120 + long_only_above_slow: false + slope_lookback: 8 + cooldown: 24 + atr_sl: 1.2 + atr_tp: 3.5 + atr_window: 21 + allow_short: true + short_only_below_slow: false + rsi_period: 14 + rsi_long_thresh: 60 + rsi_short_thresh: 40 + range_mode: mean_revert + range_rsi_high: 75 + range_rsi_low: 25 + trend_min_bars: 2 + atr_percentile_min: 0.4 + htf_alignment: true + htf_rsi_range: [35, 65] + base_size_mult: 1.0 + size_tiers: + - name: strong + min_atr_pct: 0.6 + min_trend_bars: 8 + size_mult: 1.5 + - name: base + min_atr_pct: 0.45 + min_trend_strength: 0.00008 + size_mult: 1.1 + risk_rules: + - type: atr_percentile + max: 0.95 + cooldown_bars: 12 + - type: calendar + dates: ["2024-12-06", "2025-01-10"] + cooldown_bars: 24 + - name: bollinger_mean_revert + params: + window: 48 + num_std: 2.0 + enter_z: 1.4 + exit_z: 0.2 + allow_short: false + cooldown: 12 diff --git a/QuantTrader/artifacts/config/usdjpy_regime.yaml b/QuantTrader/artifacts/config/usdjpy_regime.yaml new file mode 100644 index 0000000..c3e71c5 --- /dev/null +++ b/QuantTrader/artifacts/config/usdjpy_regime.yaml @@ -0,0 +1,42 @@ +symbol: USDJPY +csv: data/raw/USDJPY_H1.csv +cash: 100000.0 +qty: 10000 +account_ccy: USD +fast: 20 +slow: 120 +spread: 1.2 +slip: 0.2 +comm: 2.0 +atr_sl: 1.5 +atr_tp: 3.0 +atr_window: 21 +cooldown: 12 +allow_short: true +long_only_above_slow: false +short_only_below_slow: false +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +regime_ema_window: 200 +regime_slope_min: 0.00015 +regime_atr_min: 0.0015 +strategies: + - name: regime_sma + params: + trend_params: + fast_win: 20 + slow_win: 120 + long_only_above_slow: false + slope_lookback: 5 + cooldown: 12 + atr_sl: 1.5 + atr_tp: 3.0 + atr_window: 21 + allow_short: true + short_only_below_slow: false + rsi_period: 14 + rsi_long_thresh: 55 + rsi_short_thresh: 45 + range_mode: mean_revert + range_rsi_high: 65 + range_rsi_low: 35 diff --git a/QuantTrader/artifacts/config/usdjpy_sharpe239.yaml b/QuantTrader/artifacts/config/usdjpy_sharpe239.yaml new file mode 100644 index 0000000..ea24757 --- /dev/null +++ b/QuantTrader/artifacts/config/usdjpy_sharpe239.yaml @@ -0,0 +1,58 @@ +symbol: USDJPY +csv: data/raw/USDJPY_H1.csv +cash: 100000.0 +qty: 10000 +account_ccy: USD +fast: 20 +slow: 120 +spread: 1.2 +slip: 0.2 +comm: 2.0 +atr_sl: 1.0 +atr_tp: 3.0 +atr_window: 21 +cooldown: 36 +allow_short: true +long_only_above_slow: false +short_only_below_slow: false +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +regime_ema_window: 200 +regime_slope_min: 0.00015 +regime_atr_min: 0.0015 +regime_atr_percentile_min: 0.4 +regime_atr_percentile_window: 400 +regime_trend_min_bars: 2 +htf_factor: 4 +htf_ema_window: 60 +htf_rsi_period: 14 +strategies: + - name: regime_sma + params: + trend_params: + fast_win: 20 + slow_win: 120 + long_only_above_slow: false + slope_lookback: 5 + cooldown: 36 + atr_sl: 1.0 + atr_tp: 3.0 + atr_window: 21 + allow_short: true + short_only_below_slow: false + rsi_period: 14 + rsi_long_thresh: 60 + rsi_short_thresh: 40 + range_mode: mean_revert + range_rsi_high: 75 + range_rsi_low: 25 + trend_min_bars: 2 + atr_percentile_min: 0.4 + base_size_mult: 1.0 + size_tiers: + - name: strong + min_atr_pct: 0.6 + min_trend_bars: 8 + size_mult: 1.5 + - name: base + size_mult: 1.0 diff --git a/QuantTrader/artifacts/params/best_params_grid.json b/QuantTrader/artifacts/params/best_params_grid.json new file mode 100644 index 0000000..e5a50c1 --- /dev/null +++ b/QuantTrader/artifacts/params/best_params_grid.json @@ -0,0 +1,13 @@ +{ + "fast": 30.0, + "slow": 100.0, + "final_equity": 100836.47454711428, + "ann_return": 0.00816107907030128, + "ann_vol": 0.006608576787318331, + "sharpe": 1.2349223339527744, + "max_drawdown": -0.003808821586655787, + "trades": 168.0, + "atr_sl": 1.5, + "atr_tp": NaN, + "atr_window": 14.0 +} \ No newline at end of file diff --git a/QuantTrader/artifacts/params/best_params_grid_USDJPY.json b/QuantTrader/artifacts/params/best_params_grid_USDJPY.json new file mode 100644 index 0000000..4cb8f82 --- /dev/null +++ b/QuantTrader/artifacts/params/best_params_grid_USDJPY.json @@ -0,0 +1,13 @@ +{ + "fast": 20.0, + "slow": 100.0, + "final_equity": 100096.73588845377, + "ann_return": 0.0009612082650605203, + "ann_vol": 0.006298998439600644, + "sharpe": 0.15259700002742987, + "max_drawdown": -0.00570671952352962, + "trades": 160.0, + "atr_sl": 1.5, + "atr_tp": NaN, + "atr_window": 14.0 +} \ No newline at end of file diff --git a/QuantTrader/artifacts/params/usdjpy_sharpe239.json b/QuantTrader/artifacts/params/usdjpy_sharpe239.json new file mode 100644 index 0000000..2d16098 --- /dev/null +++ b/QuantTrader/artifacts/params/usdjpy_sharpe239.json @@ -0,0 +1,27 @@ +{ + "symbol": "USDJPY", + "fast": 20, + "slow": 120, + "atr_sl": 1.0, + "atr_tp": 3.0, + "atr_window": 21, + "cooldown": 36, + "rsi_long_thresh": 60, + "rsi_short_thresh": 40, + "spread": 1.2, + "slip": 0.2, + "comm": 2.0, + "risk_per_trade_pct": 0.01, + "max_drawdown_pct": 0.05, + "result_summary": { + "sharpe": 2.3922795335924505, + "ann_return": 0.003221386151178729, + "ann_vol": 0.001346575977407298, + "max_drawdown": -0.0008654020935744954, + "trades": 188, + "expectancy": 2.8389997613201357, + "win_rate": 0.3617021276595745, + "median_hold": "0 days 07:00:00", + "source": "QuantResearch/data/grid/grid_rsi_trailing_diagnostics_USDJPY_grid.csv" + } +} diff --git a/QuantTrader/bin/post_session.sh b/QuantTrader/bin/post_session.sh new file mode 100644 index 0000000..ddf5412 --- /dev/null +++ b/QuantTrader/bin/post_session.sh @@ -0,0 +1,30 @@ +#!/usr/bin/env bash +set -euo pipefail +ROOT=$(cd "$(dirname "$0")/.." && pwd) +PAPER=$ROOT/results/execution/paper/fills.csv +LIVE=$ROOT/results/execution/live/fills.csv +TCA_OUT=$ROOT/results/execution/tca_summary.json +RUN_ID=${1:-$(date -u +"%Y%m%d_%H%M%S")} +if [ ! -s "$PAPER" ] || [ ! -s "$LIVE" ]; then + echo "Missing fills CSVs: $PAPER or $LIVE" >&2 + exit 1 +fi +cd "$ROOT" +python ../QuantResearch/scripts/compare_fills.py --paper "$PAPER" --live "$LIVE" --out "$TCA_OUT" +python ../QuantResearch/scripts/update_metrics_from_tca.py \ + --tca "$TCA_OUT" \ + --metrics ../QuantResearch/results/risk/metrics.csv \ + --run-id "$RUN_ID" \ + --status pass \ + --latency-avg ${LATENCY_AVG:-30} \ + --latency-p95 ${LATENCY_P95:-45} \ + --total-pnl ${TOTAL_PNL:-0} \ + --max-exposure ${MAX_EXPOSURE:-500000} \ + --max-drawdown ${MAX_DRAWDOWN:-0.05} \ + --rolling-sharpe ${ROLLING_SHARPE:-1.4} \ + --live-drawdown ${LIVE_DRAWDOWN:-0.05} \ + --live-latency-p95 ${LIVE_LATENCY_P95:-45} \ + --slippage-bps ${SLIPPAGE_BPS:-2} +cd ../QuantResearch +python scripts/watch_ops_metrics.py +source ../.env.demo && python scripts/export_metrics_prom.py --csv results/risk/metrics.csv --job risk_sim | curl --data-binary @- "$PUSHGATEWAY_URL/metrics/job/risk_sim" diff --git a/QuantTrader/config/eurusd_multi_strategy.yaml b/QuantTrader/config/eurusd_multi_strategy.yaml new file mode 100644 index 0000000..e045318 --- /dev/null +++ b/QuantTrader/config/eurusd_multi_strategy.yaml @@ -0,0 +1,33 @@ +symbol: EURUSD +csv: QuantResearch/data/raw/EURUSD_H1.csv +cash: 100000 +qty: 10000 +account_ccy: USD +fast: 20 +slow: 80 +spread: 2.0 +slip: 0.3 +comm: 0.25 +atr_sl: 1.5 +atr_tp: 3.0 +atr_window: 14 +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +allow_short: true +strategies: + - name: ma_crossover + weight: 0.6 + params: + size_mult: 1.0 + cooldown_bars: 6 + exit_buffer_pct: 0.0005 + allow_short: true + - name: momentum_breakout + weight: 0.4 + params: + lookback: 24 + enter_threshold: 0.0015 + exit_threshold: 0.0006 + size_mult: 0.8 + allow_short: true + cooldown_bars: 12 diff --git a/QuantTrader/config/risk_limits.yaml b/QuantTrader/config/risk_limits.yaml new file mode 100644 index 0000000..a54fbf8 --- /dev/null +++ b/QuantTrader/config/risk_limits.yaml @@ -0,0 +1,22 @@ +global: + starting_equity: 50000 + limits: + max_position_notional: 200000 + max_gross_leverage: 3.0 + max_daily_loss: 5000 + max_drawdown: 0.1 +strategies: + sma_atr: + starting_equity: 30000 + limits: + max_position_notional: 80000 + max_gross_leverage: 1.5 + max_daily_loss: 3000 + max_drawdown: 0.08 + bollinger: + starting_equity: 20000 + limits: + max_position_notional: 60000 + max_gross_leverage: 1.2 + max_daily_loss: 2000 + max_drawdown: 0.05 diff --git a/QuantTrader/config/risk_limits_sim.yaml b/QuantTrader/config/risk_limits_sim.yaml new file mode 100644 index 0000000..8a1c29c --- /dev/null +++ b/QuantTrader/config/risk_limits_sim.yaml @@ -0,0 +1,28 @@ +global: + starting_equity: 50000 + fast: 10 + slow: 30 + rsi_long_thresh: 45 + rsi_short_thresh: 55 + qty: 5000 + limits: + # Wide limits for simulation/data gating only. + max_position_notional: 10000000 + max_gross_leverage: 1000.0 + max_daily_loss: 5000 + max_drawdown: 0.1 +strategies: + sma_atr: + starting_equity: 30000 + limits: + max_position_notional: 80000 + max_gross_leverage: 1.5 + max_daily_loss: 3000 + max_drawdown: 0.08 + bollinger: + starting_equity: 20000 + limits: + max_position_notional: 60000 + max_gross_leverage: 1.2 + max_daily_loss: 2000 + max_drawdown: 0.05 diff --git a/QuantTrader/config/usdjpy_multi_strategy.yaml b/QuantTrader/config/usdjpy_multi_strategy.yaml new file mode 100644 index 0000000..912ca52 --- /dev/null +++ b/QuantTrader/config/usdjpy_multi_strategy.yaml @@ -0,0 +1,46 @@ +symbol: USDJPY +csv: QuantResearch/data/raw/USDJPY_H1_full.csv + +cash: 100000 +qty: 10000 +account_ccy: USD + +fast: 20 +slow: 80 +spread: 2.0 +slip: 0.3 +comm: 0.25 +atr_sl: 1.5 +atr_tp: 3.0 +atr_window: 14 +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.05 +allow_short: true + +strategy_mode: weighted +strategy_vote_threshold: 0.0 + +strategies: + - name: xgb_signal + weight: 0.4 + params: + prob_long: 0.64 + prob_exit: 0.50 + cooldown_bars: 8 + size_mult: 1.0 + - name: ma_crossover + weight: 0.35 + params: + size_mult: 1.0 + cooldown_bars: 8 + exit_buffer_pct: 0.0004 + allow_short: true + - name: momentum_breakout + weight: 0.25 + params: + lookback: 36 + enter_threshold: 0.0012 + exit_threshold: 0.0005 + size_mult: 1.0 + allow_short: true + cooldown_bars: 10 diff --git a/QuantTrader/config/usdjpy_xgb.yaml b/QuantTrader/config/usdjpy_xgb.yaml new file mode 100644 index 0000000..c5e0777 --- /dev/null +++ b/QuantTrader/config/usdjpy_xgb.yaml @@ -0,0 +1,38 @@ +symbol: USDJPY +csv: QuantResearch/data/raw/USDJPY_H1_full.csv + +# Trading capital and unit size +cash: 100000.0 +qty: 10000 +account_ccy: USD + +# Cost assumptions (aligned with training defaults) +spread: 2.0 +slip: 0.3 +comm: 0.25 + +# Core engine params (still used for indicators and safety exits) +fast: 20 +slow: 80 +atr_sl: 1.5 +atr_tp: 3.0 +atr_window: 14 +regime_ema_window: 200 +cooldown: 0 + +# Risk controls +risk_per_trade_pct: 0.01 +max_drawdown_pct: 0.03 +allow_short: false + +# Single-model (long-only) strategy +strategies: + - name: xgb_signal + weight: 1.0 + params: + # If omitted, strategy will read QuantResearch/artifacts/models/usdjpy_h1_xgb_latest.json + # model_dir: QuantResearch/artifacts/models/usdjpy_h1_xgb/20250101_120000 + prob_long: 0.64 + prob_exit: 0.50 + size_mult: 1.0 + cooldown_bars: 8 diff --git a/QuantTrader/core/__init__.py b/QuantTrader/core/__init__.py new file mode 100644 index 0000000..792ec5e --- /dev/null +++ b/QuantTrader/core/__init__.py @@ -0,0 +1,10 @@ +""" +QuantTrader core package. + +Expose runtime modules (data, execution, risk, strategy). +""" + +from . import data # noqa: F401 +from . import execution # noqa: F401 +from . import risk # noqa: F401 +from . import strategy # noqa: F401 diff --git a/QuantTrader/core/__pycache__/__init__.cpython-312.pyc b/QuantTrader/core/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..11a1e05 Binary files /dev/null and b/QuantTrader/core/__pycache__/__init__.cpython-312.pyc differ diff --git a/QuantTrader/core/__pycache__/__init__.cpython-313.pyc b/QuantTrader/core/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..99c3425 Binary files /dev/null and b/QuantTrader/core/__pycache__/__init__.cpython-313.pyc differ diff --git a/QuantTrader/core/__pycache__/events.cpython-312.pyc b/QuantTrader/core/__pycache__/events.cpython-312.pyc new file mode 100644 index 0000000..0d799b5 Binary files /dev/null and b/QuantTrader/core/__pycache__/events.cpython-312.pyc differ diff --git a/QuantTrader/core/__pycache__/events.cpython-313.pyc b/QuantTrader/core/__pycache__/events.cpython-313.pyc new file mode 100644 index 0000000..6fd2c1a Binary files /dev/null and b/QuantTrader/core/__pycache__/events.cpython-313.pyc differ diff --git a/QuantTrader/core/__pycache__/oanda_execution.cpython-312.pyc b/QuantTrader/core/__pycache__/oanda_execution.cpython-312.pyc new file mode 100644 index 0000000..f87ffc8 Binary files /dev/null and b/QuantTrader/core/__pycache__/oanda_execution.cpython-312.pyc differ diff --git a/QuantTrader/core/__pycache__/oanda_execution.cpython-313.pyc b/QuantTrader/core/__pycache__/oanda_execution.cpython-313.pyc new file mode 100644 index 0000000..632bfd9 Binary files /dev/null and b/QuantTrader/core/__pycache__/oanda_execution.cpython-313.pyc differ diff --git a/QuantTrader/core/data/__init__.py b/QuantTrader/core/data/__init__.py new file mode 100644 index 0000000..9fa0209 --- /dev/null +++ b/QuantTrader/core/data/__init__.py @@ -0,0 +1,3 @@ +# core.data package init +from . import base +from . import oanda diff --git a/QuantTrader/core/data/__pycache__/__init__.cpython-312.pyc b/QuantTrader/core/data/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..532c5f9 Binary files /dev/null and b/QuantTrader/core/data/__pycache__/__init__.cpython-312.pyc differ diff --git a/QuantTrader/core/data/__pycache__/__init__.cpython-313.pyc b/QuantTrader/core/data/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..2c135df Binary files /dev/null and b/QuantTrader/core/data/__pycache__/__init__.cpython-313.pyc differ diff --git a/QuantTrader/core/data/__pycache__/base.cpython-312.pyc b/QuantTrader/core/data/__pycache__/base.cpython-312.pyc new file mode 100644 index 0000000..799570b Binary files /dev/null and b/QuantTrader/core/data/__pycache__/base.cpython-312.pyc differ diff --git a/QuantTrader/core/data/__pycache__/base.cpython-313.pyc b/QuantTrader/core/data/__pycache__/base.cpython-313.pyc new file mode 100644 index 0000000..4d030e7 Binary files /dev/null and b/QuantTrader/core/data/__pycache__/base.cpython-313.pyc differ diff --git a/QuantTrader/core/data/__pycache__/oanda.cpython-312.pyc b/QuantTrader/core/data/__pycache__/oanda.cpython-312.pyc new file mode 100644 index 0000000..02b9b37 Binary files /dev/null and b/QuantTrader/core/data/__pycache__/oanda.cpython-312.pyc differ diff --git a/QuantTrader/core/data/__pycache__/oanda.cpython-313.pyc b/QuantTrader/core/data/__pycache__/oanda.cpython-313.pyc new file mode 100644 index 0000000..5033ed8 Binary files /dev/null and b/QuantTrader/core/data/__pycache__/oanda.cpython-313.pyc differ diff --git a/QuantTrader/core/data/base.py b/QuantTrader/core/data/base.py new file mode 100644 index 0000000..36f9fc8 --- /dev/null +++ b/QuantTrader/core/data/base.py @@ -0,0 +1,136 @@ +from abc import ABC, abstractmethod +from typing import Dict, List, Optional, Any +from datetime import datetime +import pandas as pd + +class DataFeed(ABC): + """ + 数据源的基础抽象类 + 定义了获取市场数据的标准接口 + """ + + def __init__(self, instrument: str, timeframe: str): + """ + 初始化数据源 + + Args: + instrument: 交易品种 (例如: "EUR_USD") + timeframe: 时间周期 (例如: "H1", "D") + """ + self.instrument = instrument + self.timeframe = timeframe + + @abstractmethod + async def get_historical_data( + self, + start: datetime, + end: datetime, + **kwargs + ) -> pd.DataFrame: + """ + 获取历史数据 + + Args: + start: 开始时间 + end: 结束时间 + **kwargs: 额外参数 + + Returns: + 包含历史数据的DataFrame，至少应该包含以下列： + - datetime: 时间戳 + - open: 开盘价 + - high: 最高价 + - low: 最低价 + - close: 收盘价 + - volume: 成交量（如果可用） + """ + pass + + @abstractmethod + async def get_latest_data(self) -> Dict[str, Any]: + """ + 获取最新的市场数据 + + Returns: + 包含最新市场数据的字典 + """ + pass + + @abstractmethod + async def subscribe(self, callback) -> None: + """ + 订阅实时数据更新 + + Args: + callback: 处理实时数据的回调函数 + """ + pass + + @abstractmethod + async def unsubscribe(self) -> None: + """ + 取消订阅实时数据 + """ + pass + +class DataProcessor: + """ + 数据处理器基类 + 用于对原始市场数据进行预处理和计算指标 + """ + + def __init__(self): + self.indicators = {} + + def add_indicator(self, name: str, func, **params): + """ + 添加技术指标计算 + + Args: + name: 指标名称 + func: 计算指标的函数 + **params: 指标参数 + """ + self.indicators[name] = { + 'function': func, + 'params': params + } + + def process_data(self, data: pd.DataFrame) -> pd.DataFrame: + """ + 处理数据并计算所有已注册的指标 + + Args: + data: 原始市场数据 + + Returns: + 添加了技术指标的DataFrame + """ + result = data.copy() + for name, indicator in self.indicators.items(): + try: + result[name] = indicator['function']( + data, + **indicator['params'] + ) + except Exception as e: + print(f"计算指标 {name} 时发生错误: {str(e)}") + return result + +class MarketDataEvent: + """ + 市场数据事件类 + 用于在系统各层之间传递市场数据更新 + """ + + def __init__( + self, + instrument: str, + timestamp: datetime, + data: Dict[str, Any], + event_type: str = "MARKET_DATA" + ): + self.event_type = event_type + self.instrument = instrument + self.timestamp = timestamp + self.data = data \ No newline at end of file diff --git a/QuantTrader/core/data/oanda.py b/QuantTrader/core/data/oanda.py new file mode 100644 index 0000000..a70a761 --- /dev/null +++ b/QuantTrader/core/data/oanda.py @@ -0,0 +1,257 @@ +from __future__ import annotations + +import asyncio +import time +from datetime import datetime +from typing import Dict, Any, Optional, Sequence +import pandas as pd +from queue import Queue +import threading +from loguru import logger + +from oandapyV20 import API +try: + from oandapyV20.endpoints.pricing import PricingStream as OandaPricingStream +except ImportError: # pragma: no cover + OandaPricingStream = None # type: ignore + +from .base import DataFeed, MarketDataEvent + +def _normalize_instrument(symbol: str) -> str: + """规范化交易品种名称""" + s = symbol.upper().replace(" ", "").replace("/", "_").replace("-", "_") + if "_" in s and len(s) == 7: + return s + stripped = s.replace("_", "") + if len(stripped) == 6: + return f"{stripped[:3]}_{stripped[3:]}" + return s + +class OANDADataFeed(DataFeed): + """ + OANDA数据源实现 + 提供实时和历史市场数据 + """ + + def __init__( + self, + instrument: str, + timeframe: str, + account_id: str, + access_token: str, + environment: str = "practice", + reconnect_wait: float = 5.0, + log_heartbeat: bool = False + ): + super().__init__(instrument, timeframe) + self.account_id = account_id + self.access_token = access_token + self.environment = environment + self.reconnect_wait = reconnect_wait + self.log_heartbeat = log_heartbeat + + self.client = API(access_token=access_token, environment=environment) + self._stop = threading.Event() + self._thread: Optional[threading.Thread] = None + self._callback = None + self._latest_data = None + self._loop: Optional[asyncio.AbstractEventLoop] = None + + async def get_historical_data( + self, + start: datetime, + end: datetime, + **kwargs + ) -> pd.DataFrame: + """获取历史数据 + + Args: + start: 开始时间 + end: 结束时间 + **kwargs: 额外参数，支持： + - granularity: str, 时间周期 (如 "H1", "D") + - count: int, 返回的K线数量 + + Returns: + DataFrame包含以下列：datetime, open, high, low, close, volume + """ + from oandapyV20 import API + import oandapyV20.endpoints.instruments as instruments + + granularity = kwargs.get("granularity", self.timeframe) + price = kwargs.get("price", "M") + + # OANDA 的单次请求有最大返回数限制（例如 5000 candles），对长区间需要分段请求 + MAX_CANDLES = 5000 + + # 估算每个candle的时间长度（秒），支持常见的granularity + def granularity_seconds(g: str) -> int: + if g.endswith('H'): + return int(g[:-1]) * 3600 + if g.endswith('D'): + return int(g[:-1]) * 86400 if g[:-1].isdigit() else 86400 + if g.endswith('M') and len(g) > 1 and g[0].isdigit(): + # 例如 M1, M5 (分钟) + return int(g[1:]) * 60 if g[0] == 'M' else 30 + # 默认按小时处理 + return 3600 + + step_seconds = granularity_seconds(granularity) * MAX_CANDLES + + all_data = [] + current_start = pd.to_datetime(start) + end_ts = pd.to_datetime(end) + + while current_start < end_ts: + current_end = current_start + pd.Timedelta(seconds=step_seconds) + if current_end > end_ts: + current_end = end_ts + + params = { + "from": current_start.strftime("%Y-%m-%dT%H:%M:%S.000000Z"), + "to": current_end.strftime("%Y-%m-%dT%H:%M:%S.000000Z"), + "granularity": granularity, + "price": price + } + + request = instruments.InstrumentsCandles( + instrument=self.instrument, + params=params + ) + + try: + response = self.client.request(request) + candles = response.get("candles", []) + + for candle in candles: + if candle.get("complete"): + all_data.append({ + "datetime": pd.to_datetime(candle["time"]), + "open": float(candle["mid"]["o"]), + "high": float(candle["mid"]["h"]), + "low": float(candle["mid"]["l"]), + "close": float(candle["mid"]["c"]), + "volume": int(candle.get("volume", 0)) + }) + + except Exception as e: + logger.error(f"获取历史数据失败: {str(e)}") + raise + + # 推进起点 + current_start = current_end + + if not all_data: + return pd.DataFrame() + + df = pd.DataFrame(all_data) + df.drop_duplicates(subset=["datetime"], inplace=True) + df.sort_values(by="datetime", inplace=True) + df.set_index("datetime", inplace=True) + return df + + async def get_latest_data(self) -> Dict[str, Any]: + """获取最新数据""" + return self._latest_data if self._latest_data else {} + + async def subscribe(self, callback) -> None: + """订阅实时数据""" + self._callback = callback + self._loop = asyncio.get_running_loop() + if not self._thread or not self._thread.is_alive(): + self._start_stream() + + async def unsubscribe(self) -> None: + """取消订阅""" + self._stop.set() + if self._thread: + self._thread.join(timeout=2.0) + self._loop = None + logger.info("[OANDA] Pricing stream stopped.") + + def _start_stream(self) -> None: + """启动价格流""" + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread(target=self._run_stream, daemon=True) + self._thread.start() + logger.info( + f"[OANDA] Pricing stream started for {self.instrument} (account={self.account_id})" + ) + + def _run_stream(self) -> None: + """运行价格流""" + if OandaPricingStream is None: + raise RuntimeError( + "oandapyV20.endpoints.pricing.PricingStream is unavailable. " + "Ensure oandapyV20 is installed." + ) + params = {"instruments": self.instrument} + while not self._stop.is_set(): + request = OandaPricingStream( + accountID=self.account_id, + params=params + ) + try: + for msg in self.client.request(request): + if self._stop.is_set(): + break + self._handle_msg(msg) + except Exception as exc: + if self._stop.is_set(): + break + logger.warning( + f"[OANDA] Pricing stream error: {exc}. " + f"Reconnecting in {self.reconnect_wait}s" + ) + time.sleep(self.reconnect_wait) + + def _handle_msg(self, msg: dict) -> None: + """处理价格消息""" + msg_type = msg.get("type") + if msg_type == "HEARTBEAT": + if self.log_heartbeat: + logger.debug(f"[OANDA] Heartbeat {msg.get('time')}") + return + + if msg_type != "PRICE": + logger.debug(f"[OANDA] Skip message type={msg_type}") + return + + try: + bids = msg.get("bids") + asks = msg.get("asks") + if not bids or not asks: + return + + bid = float(bids[0]["price"]) + ask = float(asks[0]["price"]) + timestamp = pd.to_datetime(msg["time"]).to_pydatetime() + + self._latest_data = { + "bid": bid, + "ask": ask, + "timestamp": timestamp, + "mid": (bid + ask) / 2 + } + + if self._callback and self._loop: + event = MarketDataEvent( + instrument=self.instrument, + timestamp=timestamp, + data=self._latest_data + ) + try: + coro = self._callback(event) + if asyncio.iscoroutine(coro): + asyncio.run_coroutine_threadsafe(coro, self._loop) + else: + self._loop.call_soon_threadsafe(self._callback, event) + except RuntimeError as exc: + logger.warning(f"[OANDA] Failed to dispatch callback: {exc}") + elif self._callback and not self._loop: + logger.warning("[OANDA] Callback set but event loop missing; dropping tick") + + except Exception as exc: + logger.warning(f"[OANDA] Malformed price message: {msg} ({exc})") diff --git a/QuantTrader/core/events.py b/QuantTrader/core/events.py new file mode 100644 index 0000000..984257a --- /dev/null +++ b/QuantTrader/core/events.py @@ -0,0 +1,38 @@ +# fx_backtest/core/events.py +import os +import sys +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from dataclasses import dataclass +from datetime import datetime +from typing import Literal, Optional + +@dataclass(frozen=True) +class TickEvent: + ts: datetime + symbol: str + bid: float + ask: float + +@dataclass(frozen=True) +class SignalEvent: + ts: datetime + symbol: str + direction: Literal["LONG", "SHORT", "EXIT"] + size: float # 单位：合约单位/手，随你定义 + +@dataclass(frozen=True) +class OrderEvent: + ts: datetime + symbol: str + side: Literal["BUY", "SELL"] + size: float + price: Optional[float] = None # 市价可为 None；限价时填价格 + +@dataclass(frozen=True) +class FillEvent: + ts: datetime + symbol: str + side: Literal["BUY", "SELL"] + size: float + price: float + commission: float \ No newline at end of file diff --git a/QuantTrader/core/execution/__init__.py b/QuantTrader/core/execution/__init__.py new file mode 100644 index 0000000..c73ef6b --- /dev/null +++ b/QuantTrader/core/execution/__init__.py @@ -0,0 +1,14 @@ +# 执行层包初始化文件 +from .base import ExecutionHandler, OrderEvent, FillEvent +try: + from .oanda_handler import OANDAExecutionHandler +except Exception: + # optional: OANDA handler may not be available if dependencies missing + OANDAExecutionHandler = None + +__all__ = [ + "ExecutionHandler", + "OrderEvent", + "FillEvent", + "OANDAExecutionHandler", +] \ No newline at end of file diff --git a/QuantTrader/core/execution/__pycache__/__init__.cpython-312.pyc b/QuantTrader/core/execution/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..6cfc1b4 Binary files /dev/null and b/QuantTrader/core/execution/__pycache__/__init__.cpython-312.pyc differ diff --git a/QuantTrader/core/execution/__pycache__/__init__.cpython-313.pyc b/QuantTrader/core/execution/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..eacdd7a Binary files /dev/null and b/QuantTrader/core/execution/__pycache__/__init__.cpython-313.pyc differ diff --git a/QuantTrader/core/execution/__pycache__/base.cpython-312.pyc b/QuantTrader/core/execution/__pycache__/base.cpython-312.pyc new file mode 100644 index 0000000..a5c151a Binary files /dev/null and b/QuantTrader/core/execution/__pycache__/base.cpython-312.pyc differ diff --git a/QuantTrader/core/execution/__pycache__/base.cpython-313.pyc b/QuantTrader/core/execution/__pycache__/base.cpython-313.pyc new file mode 100644 index 0000000..a2c970b Binary files /dev/null and b/QuantTrader/core/execution/__pycache__/base.cpython-313.pyc differ diff --git a/QuantTrader/core/execution/__pycache__/oanda_handler.cpython-312.pyc b/QuantTrader/core/execution/__pycache__/oanda_handler.cpython-312.pyc new file mode 100644 index 0000000..14fc7c4 Binary files /dev/null and b/QuantTrader/core/execution/__pycache__/oanda_handler.cpython-312.pyc differ diff --git a/QuantTrader/core/execution/__pycache__/oanda_handler.cpython-313.pyc b/QuantTrader/core/execution/__pycache__/oanda_handler.cpython-313.pyc new file mode 100644 index 0000000..4277fa2 Binary files /dev/null and b/QuantTrader/core/execution/__pycache__/oanda_handler.cpython-313.pyc differ diff --git a/QuantTrader/core/execution/base.py b/QuantTrader/core/execution/base.py new file mode 100644 index 0000000..1af2e4c --- /dev/null +++ b/QuantTrader/core/execution/base.py @@ -0,0 +1,182 @@ +from abc import ABC, abstractmethod +from typing import Dict, List, Optional, Any +from datetime import datetime +import asyncio +from loguru import logger + +from ..strategy.base import SignalEvent, Position + +class OrderEvent: + """订单事件""" + + def __init__( + self, + instrument: str, + order_type: str, # "MARKET", "LIMIT", "STOP" + direction: str, # "BUY", "SELL" + quantity: float, + timestamp: datetime, + price: Optional[float] = None, + stop_loss: Optional[float] = None, + take_profit: Optional[float] = None, + order_id: Optional[str] = None + ): + self.event_type = "ORDER" + self.instrument = instrument + self.order_type = order_type + self.direction = direction + self.quantity = quantity + self.timestamp = timestamp + self.price = price + self.stop_loss = stop_loss + self.take_profit = take_profit + self.order_id = order_id + self.status = "CREATED" # CREATED, SUBMITTED, FILLED, CANCELLED, REJECTED + +class FillEvent: + """成交事件""" + + def __init__( + self, + instrument: str, + direction: str, + quantity: float, + price: float, + timestamp: datetime, + commission: float = 0.0, + order_id: Optional[str] = None + ): + self.event_type = "FILL" + self.instrument = instrument + self.direction = direction + self.quantity = quantity + self.price = price + self.timestamp = timestamp + self.commission = commission + self.order_id = order_id + +class ExecutionHandler(ABC): + """ + 执行处理器基类 + 负责订单执行和管理 + """ + + def __init__(self): + self.orders: Dict[str, OrderEvent] = {} # order_id -> OrderEvent + self.positions: Dict[str, Position] = {} # instrument -> Position + self.fills: List[FillEvent] = [] + self._order_callbacks = [] + self._fill_callbacks = [] + + async def process_signal(self, signal: SignalEvent, price: Optional[float] = None) -> OrderEvent: + """ + 处理交易信号并创建订单 + + Args: + signal: 交易信号 + + Returns: + 创建的订单事件 + """ + order = OrderEvent( + instrument=signal.instrument, + order_type="MARKET", # 默认为市价单 + direction="BUY" if signal.signal_type == "LONG" else "SELL", + quantity=abs(signal.strength), + timestamp=signal.timestamp, + stop_loss=signal.stop_loss, + take_profit=signal.take_profit, + price=price + ) + + # 生成订单ID + order.order_id = f"{order.instrument}_{order.timestamp.strftime('%Y%m%d_%H%M%S')}" + self.orders[order.order_id] = order + + # 执行订单 + try: + await self.execute_order(order) + except Exception as e: + logger.error(f"订单执行失败: {str(e)}") + order.status = "REJECTED" + + return order + + @abstractmethod + async def execute_order(self, order: OrderEvent) -> None: + """ + 执行订单 + + Args: + order: 要执行的订单 + """ + pass + + @abstractmethod + async def cancel_order(self, order_id: str) -> bool: + """ + 取消订单 + + Args: + order_id: 要取消的订单ID + + Returns: + 是否成功取消 + """ + pass + + def add_order_callback(self, callback): + """添加订单状态更新回调""" + self._order_callbacks.append(callback) + + def add_fill_callback(self, callback): + """添加成交更新回调""" + self._fill_callbacks.append(callback) + + async def _notify_order(self, order: OrderEvent): + """通知订单状态更新""" + for callback in self._order_callbacks: + await callback(order) + + async def _notify_fill(self, fill: FillEvent): + """通知成交更新""" + for callback in self._fill_callbacks: + await callback(fill) + + def get_position(self, instrument: str) -> Optional[Position]: + """获取某个品种的持仓""" + return self.positions.get(instrument) + + def get_all_positions(self) -> List[Position]: + """获取所有持仓""" + return list(self.positions.values()) + + def update_position(self, fill: FillEvent) -> None: + """根据成交更新持仓""" + instrument = fill.instrument + position = self.positions.get(instrument) + + if position is None: + # 新建仓位 + position = Position( + instrument=instrument, + direction=fill.direction, + size=fill.quantity, + entry_price=fill.price, + entry_time=fill.timestamp + ) + self.positions[instrument] = position + else: + # 更新现有仓位 + if fill.direction == position.direction: + # 同向加仓 + new_size = position.size + fill.quantity + position.entry_price = (position.entry_price * position.size + + fill.price * fill.quantity) / new_size + position.size = new_size + else: + # 反向减仓 + position.size -= fill.quantity + if position.size <= 0: + # 清仓 + del self.positions[instrument] diff --git a/QuantTrader/core/execution/oanda_handler.py b/QuantTrader/core/execution/oanda_handler.py new file mode 100644 index 0000000..a95dcd7 --- /dev/null +++ b/QuantTrader/core/execution/oanda_handler.py @@ -0,0 +1,117 @@ +from typing import Optional +from datetime import datetime +from loguru import logger +from oandapyV20 import API +import oandapyV20.endpoints.orders as orders +import csv +from pathlib import Path + +from .base import ExecutionHandler, OrderEvent, FillEvent + +class OANDAExecutionHandler(ExecutionHandler): + """实盘环境下的 OANDA 执行实现。""" + + def __init__(self, account_id: str, access_token: str, environment: str = "practice", fills_path: Optional[str] = None): + super().__init__() + self.client = API(access_token=access_token, environment=environment) + self.account_id = account_id + default_dir = Path("results/execution/live") if environment == "live" else Path("results/execution/paper") + default_dir.mkdir(parents=True, exist_ok=True) + self._fills_csv = Path(fills_path) if fills_path else default_dir / "fills.csv" + if not self._fills_csv.exists(): + self._init_csv() + + def _init_csv(self) -> None: + with self._fills_csv.open("w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter( + f, + fieldnames=["order_id", "ts", "symbol", "pnl", "adapter_latency_ms", "direction", "price", "quantity"], + ) + writer.writeheader() + + async def execute_order(self, order: OrderEvent) -> None: + # 将 OrderEvent 转换为 OANDA 下单请求 + instrument = order.instrument.replace("_", "_") + units = int(order.quantity) if order.direction.upper() in ("BUY", "LONG") else -int(order.quantity) + + order_body = { + "instrument": instrument, + "units": str(units), + "timeInForce": "FOK", + "positionFill": "DEFAULT", + } + if order.price: + order_body["type"] = "LIMIT" + order_body["price"] = f"{order.price:.5f}" + order_body["timeInForce"] = "GTC" + else: + order_body["type"] = "MARKET" + + payload = {"order": order_body} + req = orders.OrderCreate(accountID=self.account_id, data=payload) + try: + resp = self.client.request(req) + except Exception as exc: + logger.error(f"[OANDA] Order submission failed for {instrument}: {exc}") + order.status = "REJECTED" + await self._notify_order(order) + return + + order.status = "SUBMITTED" + await self._notify_order(order) + + fill_txn = resp.get("orderFillTransaction") + if not fill_txn: + logger.warning(f"[OANDA] Order accepted but no fill: {resp}") + return + + try: + price = float(fill_txn["price"]) + filled_units = abs(float(fill_txn["units"])) + commission = float(fill_txn.get("commission", 0)) + ts = datetime.fromisoformat(fill_txn["time"].replace("Z", "+00:00")) + side = "BUY" if float(fill_txn["units"]) > 0 else "SELL" + except Exception as exc: + logger.error(f"[OANDA] Unable to parse fill transaction: {fill_txn} ({exc})") + return + + fill = FillEvent( + instrument=order.instrument, + direction=side, + quantity=filled_units, + price=price, + timestamp=ts, + commission=abs(commission), + order_id=order.order_id + ) + + order.status = "FILLED" + await self._notify_order(order) + await self._notify_fill(fill) + self.fills.append(fill) + self.update_position(fill) + self._append_fill_csv(fill) + + def _append_fill_csv(self, fill: FillEvent) -> None: + record = { + "order_id": fill.order_id or "", + "ts": fill.timestamp.isoformat(), + "symbol": fill.instrument, + "pnl": 0.0, + "adapter_latency_ms": None, + "direction": fill.direction, + "price": fill.price, + "quantity": fill.quantity, + } + with self._fills_csv.open("a", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=record.keys()) + writer.writerow(record) + + async def cancel_order(self, order_id: str) -> bool: + # OANDA 取消需要调用 OrderCancel 或交易 API；简单实现为更新状态 + if order_id in self.orders: + order = self.orders[order_id] + order.status = "CANCELLED" + await self._notify_order(order) + return True + return False diff --git a/QuantTrader/core/oanda_execution.py b/QuantTrader/core/oanda_execution.py new file mode 100644 index 0000000..3fc54eb --- /dev/null +++ b/QuantTrader/core/oanda_execution.py @@ -0,0 +1,93 @@ +""" +OANDA 执行适配器：把 OrderEvent 转换为 OANDA API 下单，并回写 FillEvent。 +""" + +from __future__ import annotations + +from queue import Queue + +import pandas as pd +from loguru import logger +from oandapyV20 import API +import oandapyV20.endpoints.orders as orders + +from .events import FillEvent, OrderEvent + + +def _normalize_instrument(symbol: str) -> str: + s = symbol.upper().replace(" ", "").replace("/", "_").replace("-", "_") + if "_" in s and len(s) == 7: + return s + stripped = s.replace("_", "") + if len(stripped) == 6: + return f"{stripped[:3]}_{stripped[3:]}" + return s + + +class OandaExecution: + """ + 把 OrderEvent 翻译为 OANDA 订单；成交后投递 FillEvent。 + """ + + def __init__( + self, + q: Queue, + account_id: str, + access_token: str, + environment: str = "practice", + ) -> None: + self.q = q + self.account_id = account_id + self.client = API(access_token=access_token, environment=environment) + + def on_event(self, ev) -> None: + if not isinstance(ev, OrderEvent): + return + instrument = _normalize_instrument(ev.symbol) + units = ev.size if ev.side == "BUY" else -ev.size + order_body = { + "instrument": instrument, + "units": str(int(units)), + "timeInForce": "FOK", + "positionFill": "DEFAULT", + } + if ev.price is None: + order_body["type"] = "MARKET" + else: + order_body["type"] = "LIMIT" + order_body["timeInForce"] = "GTC" + order_body["price"] = f"{ev.price:.5f}" + + payload = {"order": order_body} + req = orders.OrderCreate(accountID=self.account_id, data=payload) + try: + resp = self.client.request(req) + except Exception as exc: + logger.error(f"[OANDA] Order submission failed for {instrument}: {exc}") + return + + fill_txn = resp.get("orderFillTransaction") + if not fill_txn: + logger.warning(f"[OANDA] Order accepted but no fill: {resp}") + return + + try: + price = float(fill_txn["price"]) + filled_units = abs(float(fill_txn["units"])) + commission = float(fill_txn.get("commission", 0)) + ts = pd.to_datetime(fill_txn["time"]).to_pydatetime() + side = "BUY" if float(fill_txn["units"]) > 0 else "SELL" + except Exception as exc: + logger.error(f"[OANDA] Unable to parse fill transaction: {fill_txn} ({exc})") + return + + self.q.put( + FillEvent( + ts=ts, + symbol=ev.symbol, + side=side, + size=filled_units, + price=price, + commission=abs(commission), + ) + ) diff --git a/QuantTrader/core/risk/__init__.py b/QuantTrader/core/risk/__init__.py new file mode 100644 index 0000000..959b38c --- /dev/null +++ b/QuantTrader/core/risk/__init__.py @@ -0,0 +1,2 @@ +# core.risk package init +from . import base diff --git a/QuantTrader/core/risk/__pycache__/__init__.cpython-312.pyc b/QuantTrader/core/risk/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..3b20600 Binary files /dev/null and b/QuantTrader/core/risk/__pycache__/__init__.cpython-312.pyc differ diff --git a/QuantTrader/core/risk/__pycache__/__init__.cpython-313.pyc b/QuantTrader/core/risk/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..b08b424 Binary files /dev/null and b/QuantTrader/core/risk/__pycache__/__init__.cpython-313.pyc differ diff --git a/QuantTrader/core/risk/__pycache__/base.cpython-312.pyc b/QuantTrader/core/risk/__pycache__/base.cpython-312.pyc new file mode 100644 index 0000000..814fee0 Binary files /dev/null and b/QuantTrader/core/risk/__pycache__/base.cpython-312.pyc differ diff --git a/QuantTrader/core/risk/__pycache__/base.cpython-313.pyc b/QuantTrader/core/risk/__pycache__/base.cpython-313.pyc new file mode 100644 index 0000000..9387b92 Binary files /dev/null and b/QuantTrader/core/risk/__pycache__/base.cpython-313.pyc differ diff --git a/QuantTrader/core/risk/__pycache__/risk_engine.cpython-312.pyc b/QuantTrader/core/risk/__pycache__/risk_engine.cpython-312.pyc new file mode 100644 index 0000000..8eb1a06 Binary files /dev/null and b/QuantTrader/core/risk/__pycache__/risk_engine.cpython-312.pyc differ diff --git a/QuantTrader/core/risk/base.py b/QuantTrader/core/risk/base.py new file mode 100644 index 0000000..21537e5 --- /dev/null +++ b/QuantTrader/core/risk/base.py @@ -0,0 +1,161 @@ +from abc import ABC, abstractmethod +from typing import Dict, List, Optional, Any +from datetime import datetime +import pandas as pd + +from ..strategy.base import Position, SignalEvent + +class RiskEvent: + """风险事件""" + + def __init__( + self, + event_type: str, # "RISK_LIMIT", "STOP_LOSS", "MARGIN_CALL" etc. + instrument: str, + timestamp: datetime, + message: str, + severity: str = "WARNING", # "INFO", "WARNING", "CRITICAL" + data: Optional[Dict[str, Any]] = None + ): + self.event_type = event_type + self.instrument = instrument + self.timestamp = timestamp + self.message = message + self.severity = severity + self.data = data or {} + +class PositionSizer(ABC): + """ + 仓位管理器基类 + 负责计算每笔交易的具体仓位大小 + """ + + @abstractmethod + def calculate_position_size( + self, + signal: SignalEvent, + portfolio_value: float, + risk_per_trade: float + ) -> float: + """ + 计算交易仓位大小 + + Args: + signal: 交易信号 + portfolio_value: 当前组合总价值 + risk_per_trade: 每笔交易的风险比例 + + Returns: + 建议的仓位大小 + """ + pass + +class RiskManager(ABC): + """ + 风险管理器基类 + 负责风险控制和监控 + """ + + def __init__( + self, + max_position_size: float, + max_portfolio_risk: float, + max_drawdown: float + ): + self.max_position_size = max_position_size + self.max_portfolio_risk = max_portfolio_risk + self.max_drawdown = max_drawdown + self.current_drawdown = 0.0 + self.peak_value = 0.0 + + @abstractmethod + async def check_signal(self, signal: SignalEvent) -> bool: + """ + 检查交易信号是否符合风险控制要求 + + Args: + signal: 交易信号 + + Returns: + True if signal is acceptable, False otherwise + """ + pass + + @abstractmethod + async def check_position(self, position: Position) -> List[RiskEvent]: + """ + 检查持仓的风险状况 + + Args: + position: 当前持仓 + + Returns: + 风险事件列表 + """ + pass + + def update_drawdown(self, portfolio_value: float) -> Optional[RiskEvent]: + """ + 更新和检查回撤状况 + + Args: + portfolio_value: 当前组合价值 + + Returns: + 如果超过最大回撤限制，返回风险事件 + """ + if portfolio_value > self.peak_value: + self.peak_value = portfolio_value + self.current_drawdown = 0.0 + else: + self.current_drawdown = (self.peak_value - portfolio_value) / self.peak_value + + if self.current_drawdown > self.max_drawdown: + return RiskEvent( + event_type="MAX_DRAWDOWN_BREACH", + instrument="PORTFOLIO", + timestamp=datetime.now(), + message=f"Maximum drawdown breached: {self.current_drawdown:.2%}", + severity="CRITICAL", + data={"drawdown": self.current_drawdown} + ) + return None + +class SimpleRiskManager(RiskManager): + """ + 简单风险管理器实现 + 实现基本的风险控制功能 + """ + + async def check_signal(self, signal: SignalEvent) -> bool: + """检查交易信号""" + # 实现基本的信号检查逻辑 + if not signal.stop_loss: + return False # 要求必须有止损 + return True + + async def check_position(self, position: Position) -> List[RiskEvent]: + """检查持仓风险""" + events = [] + + # 检查持仓规模 + if abs(position.size) > self.max_position_size: + events.append(RiskEvent( + event_type="POSITION_SIZE_LIMIT", + instrument=position.instrument, + timestamp=datetime.now(), + message=f"Position size {position.size} exceeds limit {self.max_position_size}", + severity="WARNING" + )) + + # 检查止损 + if not position.stop_loss: + events.append(RiskEvent( + event_type="MISSING_STOP_LOSS", + instrument=position.instrument, + timestamp=datetime.now(), + message="Position has no stop loss", + severity="WARNING" + )) + + return events \ No newline at end of file diff --git a/QuantTrader/core/risk/risk_engine.py b/QuantTrader/core/risk/risk_engine.py new file mode 100644 index 0000000..a41d5d1 --- /dev/null +++ b/QuantTrader/core/risk/risk_engine.py @@ -0,0 +1,70 @@ +"""Lightweight risk engine enforcing exposure, leverage, and loss caps.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, Tuple + + +@dataclass +class RiskLimits: + max_position_notional: float + max_gross_leverage: float + max_daily_loss: float + max_drawdown: float + + +@dataclass +class RiskState: + equity: float = 0.0 + peak_equity: float = 0.0 + min_equity: float = float("inf") + realized_pnl: float = 0.0 + gross_notional: float = 0.0 + exposures: Dict[str, float] = field(default_factory=dict) + + +class RiskViolation(Exception): + """Raised when orders violate limits.""" + + +class RiskEngine: + def __init__(self, limits: RiskLimits, starting_equity: float): + self.limits = limits + self.state = RiskState(equity=starting_equity, peak_equity=starting_equity, min_equity=starting_equity) + + def evaluate_order(self, symbol: str, side: str, notional: float) -> Tuple[bool, str]: + exposure = self.state.exposures.get(symbol, 0.0) + proposed = exposure + (notional if side.lower() == "buy" else -notional) + if abs(proposed) > self.limits.max_position_notional: + return False, f"symbol_exposure_limit:{symbol}" + + gross = self.state.gross_notional + abs(notional) + leverage = gross / self.state.equity if self.state.equity else float("inf") + if leverage > self.limits.max_gross_leverage: + return False, "gross_leverage_limit" + return True, "ok" + + def record_fill(self, symbol: str, side: str, notional: float, pnl: float) -> None: + delta = notional if side.lower() == "buy" else -notional + self.state.exposures[symbol] = self.state.exposures.get(symbol, 0.0) + delta + self.state.gross_notional = sum(abs(v) for v in self.state.exposures.values()) + + self.state.realized_pnl += pnl + self.state.equity += pnl + self.state.peak_equity = max(self.state.peak_equity, self.state.equity) + self.state.min_equity = min(self.state.min_equity, self.state.equity) + + def check_loss_limits(self) -> Tuple[bool, str]: + if -self.state.realized_pnl > self.limits.max_daily_loss: + return False, "daily_loss_limit" + drawdown = (self.state.equity - self.state.peak_equity) / self.state.peak_equity if self.state.peak_equity else 0.0 + if drawdown < -self.limits.max_drawdown: + return False, "drawdown_limit" + return True, "ok" + + def max_drawdown_pct(self) -> float: + if not self.state.peak_equity: + return 0.0 + trough = self.state.min_equity if self.state.min_equity != float("inf") else self.state.equity + return abs((trough - self.state.peak_equity) / self.state.peak_equity) diff --git a/QuantTrader/core/strategy/__init__.py b/QuantTrader/core/strategy/__init__.py new file mode 100644 index 0000000..4095fb8 --- /dev/null +++ b/QuantTrader/core/strategy/__init__.py @@ -0,0 +1,3 @@ +# core.strategy package init +from . import base +from . import rsi_mean_reversion diff --git a/QuantTrader/core/strategy/__pycache__/__init__.cpython-312.pyc b/QuantTrader/core/strategy/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..24da7db Binary files /dev/null and b/QuantTrader/core/strategy/__pycache__/__init__.cpython-312.pyc differ diff --git a/QuantTrader/core/strategy/__pycache__/__init__.cpython-313.pyc b/QuantTrader/core/strategy/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..3b2cab6 Binary files /dev/null and b/QuantTrader/core/strategy/__pycache__/__init__.cpython-313.pyc differ diff --git a/QuantTrader/core/strategy/__pycache__/base.cpython-312.pyc b/QuantTrader/core/strategy/__pycache__/base.cpython-312.pyc new file mode 100644 index 0000000..fad23e9 Binary files /dev/null and b/QuantTrader/core/strategy/__pycache__/base.cpython-312.pyc differ diff --git a/QuantTrader/core/strategy/__pycache__/base.cpython-313.pyc b/QuantTrader/core/strategy/__pycache__/base.cpython-313.pyc new file mode 100644 index 0000000..58f4439 Binary files /dev/null and b/QuantTrader/core/strategy/__pycache__/base.cpython-313.pyc differ diff --git a/QuantTrader/core/strategy/__pycache__/rsi_mean_reversion.cpython-312.pyc b/QuantTrader/core/strategy/__pycache__/rsi_mean_reversion.cpython-312.pyc new file mode 100644 index 0000000..f65322b Binary files /dev/null and b/QuantTrader/core/strategy/__pycache__/rsi_mean_reversion.cpython-312.pyc differ diff --git a/QuantTrader/core/strategy/__pycache__/rsi_mean_reversion.cpython-313.pyc b/QuantTrader/core/strategy/__pycache__/rsi_mean_reversion.cpython-313.pyc new file mode 100644 index 0000000..1ae9a49 Binary files /dev/null and b/QuantTrader/core/strategy/__pycache__/rsi_mean_reversion.cpython-313.pyc differ diff --git a/QuantTrader/core/strategy/base.py b/QuantTrader/core/strategy/base.py new file mode 100644 index 0000000..a6e29ce --- /dev/null +++ b/QuantTrader/core/strategy/base.py @@ -0,0 +1,122 @@ +from abc import ABC, abstractmethod +from typing import Dict, List, Optional, Any +from datetime import datetime +import pandas as pd + +from ..data.base import MarketDataEvent + +class Position: + """持仓类，表示当前市场头寸""" + + def __init__( + self, + instrument: str, + direction: str, # "LONG" or "SHORT" + size: float, + entry_price: float, + entry_time: datetime, + stop_loss: Optional[float] = None, + take_profit: Optional[float] = None + ): + self.instrument = instrument + self.direction = direction + self.size = size + self.entry_price = entry_price + self.entry_time = entry_time + self.stop_loss = stop_loss + self.take_profit = take_profit + self.unrealized_pnl = 0.0 + self.realized_pnl = 0.0 + +class SignalEvent: + """交易信号事件""" + + def __init__( + self, + instrument: str, + timestamp: datetime, + signal_type: str, # "LONG", "SHORT", "EXIT" + direction: str, + strength: float = 1.0, + stop_loss: Optional[float] = None, + take_profit: Optional[float] = None + ): + self.event_type = "SIGNAL" + self.instrument = instrument + self.timestamp = timestamp + self.signal_type = signal_type + self.direction = direction + self.strength = strength + self.stop_loss = stop_loss + self.take_profit = take_profit + +class Strategy(ABC): + """ + 策略基类 + 定义了策略开发的标准接口 + """ + + def __init__( + self, + instrument: str, + position_size: float = 1.0, + max_positions: int = 1 + ): + self.instrument = instrument + self.position_size = position_size + self.max_positions = max_positions + self.positions: List[Position] = [] + self.historical_data: Optional[pd.DataFrame] = None + + @abstractmethod + async def on_data(self, event: MarketDataEvent) -> Optional[SignalEvent]: + """ + 处理市场数据更新 + + Args: + event: 市场数据事件 + + Returns: + 如果产生交易信号，返回SignalEvent；否则返回None + """ + pass + + @abstractmethod + async def calculate_signals(self, data: pd.DataFrame) -> List[SignalEvent]: + """ + 基于历史数据计算交易信号 + + Args: + data: 历史市场数据 + + Returns: + 交易信号列表 + """ + pass + + def update_position(self, position: Position, current_price: float) -> None: + """ + 更新持仓的未实现盈亏 + + Args: + position: 需要更新的持仓 + current_price: 当前市场价格 + """ + if position.direction == "LONG": + position.unrealized_pnl = (current_price - position.entry_price) * position.size + else: + position.unrealized_pnl = (position.entry_price - current_price) * position.size + + def can_open_position(self) -> bool: + """检查是否可以开新仓位""" + return len(self.positions) < self.max_positions + + def get_position_value(self) -> float: + """获取当前持仓的总价值""" + return sum(abs(pos.unrealized_pnl) for pos in self.positions) + + def get_total_pnl(self) -> float: + """获取总盈亏（已实现 + 未实现）""" + unrealized = sum(pos.unrealized_pnl for pos in self.positions) + realized = sum(pos.realized_pnl for pos in self.positions) + return realized + unrealized \ No newline at end of file diff --git a/QuantTrader/core/strategy/ma_crossover.py b/QuantTrader/core/strategy/ma_crossover.py new file mode 100644 index 0000000..4bb2e25 --- /dev/null +++ b/QuantTrader/core/strategy/ma_crossover.py @@ -0,0 +1,65 @@ +from typing import List, Optional +import pandas as pd +from datetime import datetime + +from .base import Strategy, SignalEvent +from .base import Position +from ..data.base import MarketDataEvent + +class MACrossoverStrategy(Strategy): + """ + 简单移动平均交叉策略 + 快线上穿慢线做多，下穿做空 + """ + def __init__(self, instrument: str, fast_period: int = 50, slow_period: int = 200, position_size: float = 1.0): + super().__init__(instrument, position_size) + self.fast_period = fast_period + self.slow_period = slow_period + + async def on_data(self, event: MarketDataEvent) -> Optional[SignalEvent]: + if self.historical_data is None: + return None + # 更新收盘价 + close = event.data.get('close') or event.data.get('mid') + self.historical_data.loc[event.timestamp] = { + 'open': event.data.get('open', close), + 'high': event.data.get('high', close), + 'low': event.data.get('low', close), + 'close': close + } + if len(self.historical_data) < self.slow_period: + return None + + fast = self.historical_data['close'].rolling(self.fast_period).mean() + slow = self.historical_data['close'].rolling(self.slow_period).mean() + + if fast.iloc[-2] <= slow.iloc[-2] and fast.iloc[-1] > slow.iloc[-1]: + return SignalEvent( + instrument=self.instrument, + timestamp=event.timestamp, + signal_type="LONG", + direction="BUY", + strength=self.position_size + ) + if fast.iloc[-2] >= slow.iloc[-2] and fast.iloc[-1] < slow.iloc[-1]: + return SignalEvent( + instrument=self.instrument, + timestamp=event.timestamp, + signal_type="SHORT", + direction="SELL", + strength=self.position_size + ) + return None + + async def calculate_signals(self, data: pd.DataFrame) -> List[SignalEvent]: + signals = [] + self.historical_data = data.copy() + fast = data['close'].rolling(self.fast_period).mean() + slow = data['close'].rolling(self.slow_period).mean() + for i in range(self.slow_period, len(data)): + ts = data.index[i] + if fast.iloc[i-1] <= slow.iloc[i-1] and fast.iloc[i] > slow.iloc[i]: + signals.append(SignalEvent(instrument=self.instrument, timestamp=ts, signal_type="LONG", direction="BUY", strength=self.position_size)) + if fast.iloc[i-1] >= slow.iloc[i-1] and fast.iloc[i] < slow.iloc[i]: + signals.append(SignalEvent(instrument=self.instrument, timestamp=ts, signal_type="SHORT", direction="SELL", strength=self.position_size)) + return signals diff --git a/QuantTrader/core/strategy/momentum.py b/QuantTrader/core/strategy/momentum.py new file mode 100644 index 0000000..5b9a8c4 --- /dev/null +++ b/QuantTrader/core/strategy/momentum.py @@ -0,0 +1,46 @@ +from typing import List, Optional +import pandas as pd +from datetime import datetime + +from .base import Strategy, SignalEvent +from ..data.base import MarketDataEvent + +class MomentumStrategy(Strategy): + """ + 简单动量策略：当价格高于N日均线时做多，低于时做空 + """ + def __init__(self, instrument: str, lookback: int = 20, position_size: float = 1.0): + super().__init__(instrument, position_size) + self.lookback = lookback + + async def on_data(self, event: MarketDataEvent) -> Optional[SignalEvent]: + if self.historical_data is None: + return None + close = event.data.get('close') or event.data.get('mid') + self.historical_data.loc[event.timestamp] = { + 'open': event.data.get('open', close), + 'high': event.data.get('high', close), + 'low': event.data.get('low', close), + 'close': close + } + if len(self.historical_data) < self.lookback: + return None + ma = self.historical_data['close'].rolling(self.lookback).mean() + if close > ma.iloc[-1] and self.can_open_position(): + return SignalEvent(self.instrument, event.timestamp, "LONG", "BUY", strength=self.position_size) + if close < ma.iloc[-1] and self.can_open_position(): + return SignalEvent(self.instrument, event.timestamp, "SHORT", "SELL", strength=self.position_size) + return None + + async def calculate_signals(self, data: pd.DataFrame) -> List[SignalEvent]: + signals = [] + self.historical_data = data.copy() + ma = data['close'].rolling(self.lookback).mean() + for i in range(self.lookback, len(data)): + ts = data.index[i] + price = data['close'].iloc[i] + if price > ma.iloc[i-1]: + signals.append(SignalEvent(self.instrument, ts, "LONG", "BUY", strength=self.position_size)) + elif price < ma.iloc[i-1]: + signals.append(SignalEvent(self.instrument, ts, "SHORT", "SELL", strength=self.position_size)) + return signals diff --git a/QuantTrader/core/strategy/rsi_mean_reversion.py b/QuantTrader/core/strategy/rsi_mean_reversion.py new file mode 100644 index 0000000..3776aa9 --- /dev/null +++ b/QuantTrader/core/strategy/rsi_mean_reversion.py @@ -0,0 +1,150 @@ +from typing import List, Optional +import pandas as pd +import numpy as np +from datetime import datetime + +from ..data.base import MarketDataEvent +from .base import Strategy, SignalEvent + +class RSIMeanReversionStrategy(Strategy): + """ + RSI均值回归策略 + 当RSI超买时做空，超卖时做多 + """ + + def __init__( + self, + instrument: str, + position_size: float = 1.0, + max_positions: int = 1, + rsi_period: int = 14, + overbought: float = 70.0, + oversold: float = 30.0, + stop_loss_atr: float = 2.0, + atr_period: int = 14 + ): + super().__init__(instrument, position_size, max_positions) + self.rsi_period = rsi_period + self.overbought = overbought + self.oversold = oversold + self.stop_loss_atr = stop_loss_atr + self.atr_period = atr_period + self.last_rsi = None + self.last_atr = None + + @staticmethod + def calculate_rsi(data: pd.Series, period: int = 14) -> pd.Series: + """计算RSI指标""" + delta = data.diff() + gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() + loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() + rs = gain / loss + return 100 - (100 / (1 + rs)) + + @staticmethod + def calculate_atr(data: pd.DataFrame, period: int = 14) -> pd.Series: + """计算ATR指标""" + high = data['high'] + low = data['low'] + close = data['close'] + + tr1 = high - low + tr2 = abs(high - close.shift()) + tr3 = abs(low - close.shift()) + tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) + + return tr.rolling(window=period).mean() + + async def on_data(self, event: MarketDataEvent) -> Optional[SignalEvent]: + """ + 处理实时市场数据 + + Args: + event: 市场数据事件 + + Returns: + 如果触发信号则返回SignalEvent，否则返回None + """ + if self.historical_data is None: + return None + + # 更新数据 + current_price = event.data['mid'] + self.historical_data.loc[event.timestamp] = current_price + + # 计算指标 + close_prices = self.historical_data['close'] + rsi = self.calculate_rsi(close_prices, self.rsi_period).iloc[-1] + atr = self.calculate_atr(self.historical_data, self.atr_period).iloc[-1] + + self.last_rsi = rsi + self.last_atr = atr + + # 生成信号 + if self.can_open_position(): + if rsi > self.overbought: + return SignalEvent( + instrument=self.instrument, + timestamp=event.timestamp, + signal_type="SHORT", + direction="SELL", + strength=self.position_size, + stop_loss=current_price + self.stop_loss_atr * atr + ) + elif rsi < self.oversold: + return SignalEvent( + instrument=self.instrument, + timestamp=event.timestamp, + signal_type="LONG", + direction="BUY", + strength=self.position_size, + stop_loss=current_price - self.stop_loss_atr * atr + ) + + return None + + async def calculate_signals(self, data: pd.DataFrame) -> List[SignalEvent]: + """ + 基于历史数据计算交易信号 + + Args: + data: 历史市场数据 + + Returns: + 交易信号列表 + """ + signals = [] + self.historical_data = data.copy() + + # 计算指标 + close_prices = data['close'] + rsi = self.calculate_rsi(close_prices, self.rsi_period) + atr = self.calculate_atr(data, self.atr_period) + + # 生成信号 + for i in range(self.rsi_period, len(data)): + timestamp = data.index[i] + current_price = close_prices[i] + current_rsi = rsi[i] + current_atr = atr[i] + + if current_rsi > self.overbought: + signals.append(SignalEvent( + instrument=self.instrument, + timestamp=timestamp, + signal_type="SHORT", + direction="SELL", + strength=self.position_size, + stop_loss=current_price + self.stop_loss_atr * current_atr + )) + elif current_rsi < self.oversold: + signals.append(SignalEvent( + instrument=self.instrument, + timestamp=timestamp, + signal_type="LONG", + direction="BUY", + strength=self.position_size, + stop_loss=current_price - self.stop_loss_atr * current_atr + )) + + return signals \ No newline at end of file diff --git a/QuantTrader/data/__pycache__/oanda_stream.cpython-312.pyc b/QuantTrader/data/__pycache__/oanda_stream.cpython-312.pyc new file mode 100644 index 0000000..23b35d1 Binary files /dev/null and b/QuantTrader/data/__pycache__/oanda_stream.cpython-312.pyc differ diff --git a/QuantTrader/data/__pycache__/oanda_stream.cpython-313.pyc b/QuantTrader/data/__pycache__/oanda_stream.cpython-313.pyc new file mode 100644 index 0000000..dc15d17 Binary files /dev/null and b/QuantTrader/data/__pycache__/oanda_stream.cpython-313.pyc differ diff --git a/QuantTrader/data/oanda_stream.py b/QuantTrader/data/oanda_stream.py new file mode 100644 index 0000000..f1d87e9 --- /dev/null +++ b/QuantTrader/data/oanda_stream.py @@ -0,0 +1,116 @@ +""" +OANDA 定价流适配器，用于从模拟/实盘账户拉取实时报价并投递 TickEvent。 +""" + +from __future__ import annotations + +import threading +import time +from queue import Queue +from typing import Iterable, List, Optional, Sequence + +import pandas as pd +from loguru import logger +from oandapyV20 import API +try: + from oandapyV20.endpoints.pricing import PricingStream +except ImportError: + PricingStream = None + +from core.events import TickEvent + + +def _normalize_instrument(symbol: str) -> str: + s = symbol.upper().replace(" ", "").replace("/", "_").replace("-", "_") + if "_" in s and len(s) == 7: + return s + stripped = s.replace("_", "") + if len(stripped) == 6: + return f"{stripped[:3]}_{stripped[3:]}" + return s + + +class OandaPricingStream: + """ + 使用 OANDA Pricing Stream API 推送 TickEvent。 + """ + + def __init__( + self, + q: Queue, + account_id: str, + instruments: Sequence[str], + access_token: str, + environment: str = "practice", + reconnect_wait: float = 5.0, + log_heartbeat: bool = False, + ) -> None: + self.q = q + self.account_id = account_id + self.instruments = [_normalize_instrument(sym) for sym in instruments] + self.log_heartbeat = log_heartbeat + self.client = API(access_token=access_token, environment=environment) + self.reconnect_wait = reconnect_wait + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + + def start(self) -> None: + if PricingStream is None: + raise RuntimeError( + "oandapyV20 未暴露 PricingStream（需 0.7.2+）。请执行 `pip install --upgrade oandapyV20` 后重试。" + ) + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + logger.info( + f"[OANDA] Pricing stream started for {','.join(self.instruments)} (account={self.account_id})" + ) + + def stop(self) -> None: + self._stop.set() + if self._thread: + self._thread.join(timeout=2.0) + logger.info("[OANDA] Pricing stream stopped.") + + def _run(self) -> None: + if PricingStream is None: + logger.error("无法启动价格流：缺少 PricingStream 类") + return + params = {"instruments": ",".join(self.instruments)} + while not self._stop.is_set(): + request = PricingStream(accountID=self.account_id, params=params) + try: + for msg in self.client.request(request): + if self._stop.is_set(): + break + self._handle_msg(msg) + except Exception as exc: + if self._stop.is_set(): + break + logger.warning(f"[OANDA] Pricing stream error: {exc}. Reconnecting in {self.reconnect_wait}s") + time.sleep(self.reconnect_wait) + + def _handle_msg(self, msg: dict) -> None: + msg_type = msg.get("type") + if msg_type == "HEARTBEAT": + if self.log_heartbeat: + logger.debug(f"[OANDA] Heartbeat {msg.get('time')}") + return + if msg_type != "PRICE": + logger.debug(f"[OANDA] Skip message type={msg_type}") + return + try: + symbol = msg["instrument"].replace("_", "") + bids = msg.get("bids") + asks = msg.get("asks") + if not bids or not asks: + return + bid = float(bids[0]["price"]) + ask = float(asks[0]["price"]) + ts = pd.Timestamp(msg["time"]).floor("us").to_pydatetime() + except Exception as exc: + logger.warning(f"[OANDA] Malformed price message: {msg} ({exc})") + return + self.q.put(TickEvent(ts=ts, symbol=symbol, bid=bid, ask=ask)) diff --git a/QuantTrader/execution/__pycache__/adapter.cpython-312.pyc b/QuantTrader/execution/__pycache__/adapter.cpython-312.pyc new file mode 100644 index 0000000..10152db Binary files /dev/null and b/QuantTrader/execution/__pycache__/adapter.cpython-312.pyc differ diff --git a/QuantTrader/execution/__pycache__/config_loader.cpython-312.pyc b/QuantTrader/execution/__pycache__/config_loader.cpython-312.pyc new file mode 100644 index 0000000..6d28a21 Binary files /dev/null and b/QuantTrader/execution/__pycache__/config_loader.cpython-312.pyc differ diff --git a/QuantTrader/execution/__pycache__/metrics_logger.cpython-312.pyc b/QuantTrader/execution/__pycache__/metrics_logger.cpython-312.pyc new file mode 100644 index 0000000..d2ef1d8 Binary files /dev/null and b/QuantTrader/execution/__pycache__/metrics_logger.cpython-312.pyc differ diff --git a/QuantTrader/execution/__pycache__/oanda_adapter.cpython-312.pyc b/QuantTrader/execution/__pycache__/oanda_adapter.cpython-312.pyc new file mode 100644 index 0000000..7fba5a1 Binary files /dev/null and b/QuantTrader/execution/__pycache__/oanda_adapter.cpython-312.pyc differ diff --git a/QuantTrader/execution/__pycache__/order_store.cpython-312.pyc b/QuantTrader/execution/__pycache__/order_store.cpython-312.pyc new file mode 100644 index 0000000..0bc9f02 Binary files /dev/null and b/QuantTrader/execution/__pycache__/order_store.cpython-312.pyc differ diff --git a/QuantTrader/execution/__pycache__/paper_adapter.cpython-312.pyc b/QuantTrader/execution/__pycache__/paper_adapter.cpython-312.pyc new file mode 100644 index 0000000..d45595e Binary files /dev/null and b/QuantTrader/execution/__pycache__/paper_adapter.cpython-312.pyc differ diff --git a/QuantTrader/execution/adapter.py b/QuantTrader/execution/adapter.py new file mode 100644 index 0000000..fa1240b --- /dev/null +++ b/QuantTrader/execution/adapter.py @@ -0,0 +1,102 @@ +"""Execution adapter interfaces and data models for Phase 3.""" + +from __future__ import annotations + +import abc +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Optional + + +@dataclass +class OrderParams: + symbol: str + side: str # "buy" or "sell" + quantity: float + price: Optional[float] = None + tif: str = "GTC" + client_order_id: Optional[str] = None + metadata: Optional[Dict[str, str]] = None + + +@dataclass +class OrderAck: + order_id: str + status: str # accepted/rejected + timestamp: datetime + reason: Optional[str] = None + + +@dataclass +class CancelAck: + order_id: str + status: str + timestamp: datetime + reason: Optional[str] = None + + +@dataclass +class FillEvent: + order_id: str + fill_id: str + symbol: str + side: str + quantity: float + price: float + timestamp: datetime + + +@dataclass +class PositionState: + symbol: str + quantity: float + avg_price: float + unrealized_pnl: float = 0.0 + + +class ExecutionAdapter(abc.ABC): + """Abstract interface for broker/order routing adapters.""" + + @abc.abstractmethod + def submit(self, order: OrderParams) -> OrderAck: + """Submit a new order to the venue.""" + + @abc.abstractmethod + def cancel(self, order_id: str) -> CancelAck: + """Cancel an existing order.""" + + @abc.abstractmethod + def sync_positions(self) -> Dict[str, PositionState]: + """Return latest position snapshot.""" + + @abc.abstractmethod + def heartbeat(self) -> bool: + """Quick connectivity check.""" + + +class MockAdapter(ExecutionAdapter): + """MVP mock adapter used in simulation/testing pipelines.""" + + def __init__(self): + self._order_counter = 0 + self._orders: Dict[str, OrderParams] = {} + + def _next_id(self) -> str: + self._order_counter += 1 + return f"SIM-{self._order_counter}" + + def submit(self, order: OrderParams) -> OrderAck: + order_id = self._next_id() + self._orders[order_id] = order + return OrderAck(order_id=order_id, status="accepted", timestamp=datetime.utcnow()) + + def cancel(self, order_id: str) -> CancelAck: + status = "cancelled" if order_id in self._orders else "not_found" + self._orders.pop(order_id, None) + return CancelAck(order_id=order_id, status=status, timestamp=datetime.utcnow()) + + def sync_positions(self) -> Dict[str, PositionState]: + return {} + + def heartbeat(self) -> bool: + return True diff --git a/QuantTrader/execution/config_loader.py b/QuantTrader/execution/config_loader.py new file mode 100644 index 0000000..644f40f --- /dev/null +++ b/QuantTrader/execution/config_loader.py @@ -0,0 +1,24 @@ +"""Helpers to load execution adapter configuration.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, Optional + +import yaml + + +def load_oanda_config(path: Optional[str] = None) -> Dict[str, Any]: + cfg_path = Path(path or "QuantTrader/config/execution_oanda.yaml") + if not cfg_path.exists(): + raise FileNotFoundError(f"OANDA config not found: {cfg_path}") + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) or {} + return { + "account_id": data.get("account_id"), + "base_url": data.get("base_url", "https://api-fxpractice.oanda.com/v3"), + "timeout_ms": data.get("timeout_ms", 10000), + "retry_backoff": data.get("retry_backoff", 1.0), + "max_retries": data.get("max_retries", 3), + "metrics_path": data.get("metrics_path"), + "error_log": data.get("error_log", "results/execution/errors.log"), + } diff --git a/QuantTrader/execution/metrics_logger.py b/QuantTrader/execution/metrics_logger.py new file mode 100644 index 0000000..609251f --- /dev/null +++ b/QuantTrader/execution/metrics_logger.py @@ -0,0 +1,51 @@ +"""Append execution metrics to CSV for monitoring.""" + +from __future__ import annotations + +import csv +import os +from dataclasses import asdict, dataclass +from datetime import datetime +from pathlib import Path +from typing import Optional + + +@dataclass +class ExecutionMetric: + event: str + order_id: str + symbol: str + latency_ms: Optional[float] + status: str + timestamp: str + + +class MetricsLogger: + def __init__(self, path: Optional[str] = None): + metrics_path = path or os.environ.get("EXECUTION_METRICS_PATH", "metrics/execution.csv") + self.path = Path(metrics_path) + self.path.parent.mkdir(parents=True, exist_ok=True) + if not self.path.exists(): + with self.path.open("w", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=list(ExecutionMetric.__annotations__.keys())) + writer.writeheader() + + def log(self, metric: ExecutionMetric) -> None: + with self.path.open("a", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=list(ExecutionMetric.__annotations__.keys())) + writer.writerow(asdict(metric)) + + +def log_event(event: str, order_id: str, symbol: str, status: str, start_ts: datetime, end_ts: datetime) -> None: + latency_ms = (end_ts - start_ts).total_seconds() * 1000.0 + logger = MetricsLogger() + logger.log( + ExecutionMetric( + event=event, + order_id=order_id, + symbol=symbol, + latency_ms=latency_ms, + status=status, + timestamp=end_ts.isoformat(), + ) + ) diff --git a/QuantTrader/execution/oanda_adapter.py b/QuantTrader/execution/oanda_adapter.py new file mode 100644 index 0000000..fdf9818 --- /dev/null +++ b/QuantTrader/execution/oanda_adapter.py @@ -0,0 +1,134 @@ +"""OANDA REST adapter (mock/stub for Phase 3).""" + +from __future__ import annotations + +import os +from dataclasses import asdict +from datetime import datetime +from pathlib import Path +from typing import Dict, Optional + +import requests + +from .adapter import CancelAck, ExecutionAdapter, OrderAck, OrderParams, PositionState +from .config_loader import load_oanda_config +from .metrics_logger import MetricsLogger, log_event +from .order_store import OrderStore + + +class OandaAdapter(ExecutionAdapter): + BASE_URL = "https://api-fxpractice.oanda.com/v3" + + def __init__( + self, + account_id: Optional[str] = None, + token: Optional[str] = None, + order_store: Optional[OrderStore] = None, + base_url: Optional[str] = None, + timeout_ms: int = 10000, + retry_backoff: float = 1.0, + max_retries: int = 3, + metrics_path: Optional[str] = None, + error_log: str = "results/execution/errors.log", + ) -> None: + self.account_id = account_id or os.environ.get("OANDA_ACCOUNT_ID", "") + self.token = token or os.environ.get("OANDA_TOKEN", "") + self.base_url = base_url or self.BASE_URL + self.timeout_ms = timeout_ms + self.retry_backoff = retry_backoff + self.max_retries = max_retries + self.metrics_logger = MetricsLogger(metrics_path) + self.error_log = Path(error_log) + self.session = requests.Session() + self.session.headers.update({"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}) + self.order_store = order_store or OrderStore() + + def _endpoint(self, path: str) -> str: + return f"{self.base_url}{path}" + + @classmethod + def from_config(cls, path: Optional[str] = None, order_store: Optional[OrderStore] = None) -> "OandaAdapter": + cfg = load_oanda_config(path) + return cls(order_store=order_store, **cfg) + + def _log_error(self, context: str, message: str) -> None: + self.error_log.parent.mkdir(parents=True, exist_ok=True) + with self.error_log.open("a", encoding="utf-8") as fh: + fh.write(f"[{datetime.utcnow().isoformat()}] {context}: {message}\n") + + def _request(self, method: str, url: str, **kwargs): + backoff = self.retry_backoff + for attempt in range(self.max_retries): + try: + resp = self.session.request(method, url, timeout=self.timeout_ms / 1000.0, **kwargs) + except requests.RequestException as exc: + self._log_error("network", str(exc)) + time.sleep(backoff) + backoff *= 2 + continue + if resp.status_code >= 500: + self._log_error("server_error", resp.text) + time.sleep(backoff) + backoff *= 2 + continue + if resp.status_code >= 400: + self._log_error("client_error", resp.text) + resp.raise_for_status() + resp.raise_for_status() + return resp.json() + raise RuntimeError(f"Failed {method} {url} after {self.max_retries} retries") + + def submit(self, order: OrderParams) -> OrderAck: + start = datetime.utcnow() + payload = { + "order": { + "instrument": order.symbol, + "units": int(order.quantity if order.side.lower() == "buy" else -order.quantity), + "type": "MARKET" if order.price is None else "LIMIT", + "timeInForce": order.tif.upper(), + } + } + if order.price is not None: + payload["order"]["price"] = f"{order.price:.5f}" + if order.client_order_id: + payload["order"]["clientExtensions"] = {"clientOrderID": order.client_order_id} + url = self._endpoint(f"/accounts/{self.account_id}/orders") + data = self._request("POST", url, json=payload) + order_id = data.get("orderCreateTransaction", {}).get("id", "") + if order_id: + self.order_store.append(order_id, order) + end = datetime.utcnow() + log_event("submit", order_id or "", order.symbol, "accepted", start, end) + ts = data.get("time") + timestamp = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.utcnow() + return OrderAck(order_id=order_id, status="accepted", timestamp=timestamp) + + def cancel(self, order_id: str) -> CancelAck: + start = datetime.utcnow() + url = self._endpoint(f"/accounts/{self.account_id}/orders/{order_id}/cancel") + data = self._request("PUT", url) + end = datetime.utcnow() + log_event("cancel", order_id, "", "cancelled", start, end) + ts = data.get("time") + timestamp = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.utcnow() + return CancelAck(order_id=order_id, status="cancelled", timestamp=timestamp) + + def sync_positions(self) -> Dict[str, PositionState]: + url = self._endpoint(f"/accounts/{self.account_id}/positions") + data = self._request("GET", url) + positions = {} + for pos in data.get("positions", []): + symbol = pos["instrument"] + net = float(pos["netUnrealizedPL"]) + qty = float(pos.get("long", {}).get("units", 0)) + float(pos.get("short", {}).get("units", 0)) + avg_price = float(pos.get("avgPrice", 0)) + positions[symbol] = PositionState(symbol=symbol, quantity=qty, avg_price=avg_price, unrealized_pnl=net) + return positions + + def heartbeat(self) -> bool: + url = self._endpoint("/accounts") + try: + self._request("GET", url) + return True + except Exception: + return False diff --git a/QuantTrader/execution/order_store.py b/QuantTrader/execution/order_store.py new file mode 100644 index 0000000..ac535bb --- /dev/null +++ b/QuantTrader/execution/order_store.py @@ -0,0 +1,28 @@ +"""Simple JSONL order store for audit/replay.""" + +from __future__ import annotations + +import json +from dataclasses import asdict +from pathlib import Path +from typing import Iterable + +from QuantTrader.execution.adapter import OrderParams + + +class OrderStore: + def __init__(self, path: str = "results/execution/orders.log"): + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + + def append(self, order_id: str, params: OrderParams) -> None: + record = {"order_id": order_id, **asdict(params)} + with self.path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record) + "\n") + + def load(self) -> Iterable[dict]: + if not self.path.exists(): + return [] + with self.path.open("r", encoding="utf-8") as fh: + for line in fh: + yield json.loads(line) diff --git a/QuantTrader/execution/paper_adapter.py b/QuantTrader/execution/paper_adapter.py new file mode 100644 index 0000000..2d57310 --- /dev/null +++ b/QuantTrader/execution/paper_adapter.py @@ -0,0 +1,111 @@ +"""Paper trading adapter that simulates fills with configurable latency/slippage.""" + +from __future__ import annotations + +import random +import time +from datetime import datetime +from pathlib import Path +from typing import Dict + +from .adapter import CancelAck, ExecutionAdapter, OrderAck, OrderParams, PositionState +from .metrics_logger import log_event +from .order_store import OrderStore + + +class PaperAdapter(ExecutionAdapter): + def __init__( + self, + latency_ms: float = 50.0, + slippage_pips: float = 0.1, + order_store: OrderStore | None = None, + starting_equity: float = 100000.0, + equity_log_path: str = "results/execution/paper_equity.csv", + ): + self.latency_ms = latency_ms + self.slippage_pips = slippage_pips + self.order_store = order_store or OrderStore("results/execution/paper_orders.log") + self.positions: Dict[str, PositionState] = {} + self.last_price: Dict[str, float] = {} + self.cash = starting_equity + self.equity_path = Path(equity_log_path) + self.equity_path.parent.mkdir(parents=True, exist_ok=True) + if not self.equity_path.exists(): + self.equity_path.write_text("ts,equity\n", encoding="utf-8") + self._order_seq = 0 + + def _next_id(self) -> str: + self._order_seq += 1 + return f"PAPER-{self._order_seq}" + + def submit(self, order: OrderParams) -> OrderAck: + start = datetime.utcnow() + time.sleep(self.latency_ms / 1000.0) + order_id = self._next_id() + fill_price = self._fill_price(order) + self._apply_fill(order, fill_price) + self.order_store.append(order_id, order) + end = datetime.utcnow() + log_event("submit", order_id, order.symbol, "accepted", start, end) + self._record_equity(end) + return OrderAck(order_id=order_id, status="accepted", timestamp=end) + + def cancel(self, order_id: str) -> CancelAck: + start = datetime.utcnow() + time.sleep(self.latency_ms / 2000.0) + end = datetime.utcnow() + log_event("cancel", order_id, "", "cancelled", start, end) + return CancelAck(order_id=order_id, status="cancelled", timestamp=end) + + def sync_positions(self) -> dict[str, PositionState]: + return self.positions + + def heartbeat(self) -> bool: + return True + + def _pip_value(self, symbol: str) -> float: + return 0.01 if symbol.endswith("JPY") else 0.0001 + + def _fill_price(self, order: OrderParams) -> float: + base_price = order.price or self.last_price.get(order.symbol, 1.0) + slip = self.slippage_pips * self._pip_value(order.symbol) + direction = 1 if order.side.lower() == "buy" else -1 + return base_price + direction * slip * random.choice([1, -1]) + + def _apply_fill(self, order: OrderParams, price: float) -> None: + qty = order.quantity if order.side.lower() == "buy" else -order.quantity + pos = self.positions.get(order.symbol) + if pos is None: + pos = PositionState(symbol=order.symbol, quantity=0.0, avg_price=price, unrealized_pnl=0.0) + total_qty = pos.quantity + qty + if total_qty == 0: + realized = (price - pos.avg_price) * (-qty) # closing position + self.cash += realized + self.positions.pop(order.symbol, None) + else: + if pos.quantity == 0 or (pos.quantity > 0 and qty > 0) or (pos.quantity < 0 and qty < 0): + avg = ((pos.quantity * pos.avg_price) + (qty * price)) / total_qty + pos = PositionState(symbol=order.symbol, quantity=total_qty, avg_price=avg, unrealized_pnl=0.0) + self.positions[order.symbol] = pos + else: + realized = (pos.avg_price - price) * qty * -1 + self.cash += realized + pos = PositionState(symbol=order.symbol, quantity=total_qty, avg_price=pos.avg_price, unrealized_pnl=0.0) + if total_qty == 0: + self.positions.pop(order.symbol, None) + else: + self.positions[order.symbol] = pos + notional = price * order.quantity + if qty > 0: + self.cash -= notional + else: + self.cash += notional + self.last_price[order.symbol] = price + + def _record_equity(self, timestamp: datetime) -> None: + equity = self.cash + for symbol, pos in self.positions.items(): + mark = self.last_price.get(symbol, pos.avg_price) + equity += pos.quantity * mark + with self.equity_path.open("a", encoding="utf-8") as fh: + fh.write(f"{timestamp.isoformat()},{equity}\n") diff --git a/QuantTrader/logs/oanda_trading.log b/QuantTrader/logs/oanda_trading.log new file mode 100644 index 0000000..6d26d3b --- /dev/null +++ b/QuantTrader/logs/oanda_trading.log @@ -0,0 +1,14 @@ +2025-11-04 16:30:59.195 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:34:28.561 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:35:55.938 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:35:56.617 | INFO | __main__::23 - 💰 Balance: 965985.9487 GBP +2025-11-04 16:47:20.949 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:47:27.446 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:47:34.147 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:47:41.773 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:49:19.573 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:51:07.455 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:51:12.478 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:53:59.135 | INFO | __main__::18 - ✅ Connected to OANDA +2025-11-04 16:53:59.837 | INFO | __main__::23 - 💰 Balance: 965985.9487 GBP +2025-11-04 17:14:25.105 | INFO | __main__::21 - 📈 策略累计收益: 0.0123 diff --git a/QuantTrader/requirements.txt b/QuantTrader/requirements.txt new file mode 100644 index 0000000..7d627e8 --- /dev/null +++ b/QuantTrader/requirements.txt @@ -0,0 +1,10 @@ +# requirements.txt + +oandapyV20>=0.7.2 +pandas>=2.0.0 +numpy>=1.23.0 +loguru>=0.7.0 +python-dotenv>=1.0.0 +plotly>=5.0.0 +ta>=0.10.0 +pyyaml>=6.0.0 diff --git a/QuantTrader/results/execution/live/fills.csv b/QuantTrader/results/execution/live/fills.csv new file mode 100644 index 0000000..35c9fa1 --- /dev/null +++ b/QuantTrader/results/execution/live/fills.csv @@ -0,0 +1,3 @@ +order_id,ts,symbol,pnl,adapter_latency_ms,direction,price,quantity +LIVE-1,2025-11-12T09:00:00Z,EURUSD,42.5,30,BUY,1.1593,10000 +LIVE-2,2025-11-12T10:00:00Z,EURUSD,-15.2,28,SELL,1.1591,8000 diff --git a/QuantTrader/results/execution/paper/fills.csv b/QuantTrader/results/execution/paper/fills.csv new file mode 100644 index 0000000..00992a8 --- /dev/null +++ b/QuantTrader/results/execution/paper/fills.csv @@ -0,0 +1,3 @@ +order_id,ts,symbol,pnl,adapter_latency_ms,direction,price,quantity +PAPER-1,2025-11-12T09:00:00Z,EURUSD,40.0,20,BUY,1.1593,10000 +PAPER-2,2025-11-12T10:00:00Z,EURUSD,-10.0,18,SELL,1.1591,8000 diff --git a/QuantTrader/results/execution/tca_summary.json b/QuantTrader/results/execution/tca_summary.json new file mode 100644 index 0000000..3a2ff25 --- /dev/null +++ b/QuantTrader/results/execution/tca_summary.json @@ -0,0 +1,14 @@ +{ + "paper_path": "/Users/chuan/Documents/Projects/FX_Backtest/QuantTrader/results/execution/paper/fills.csv", + "live_path": "/Users/chuan/Documents/Projects/FX_Backtest/QuantTrader/results/execution/live/fills.csv", + "paper_trade_count": 2, + "paper_total_pnl": 30.0, + "paper_avg_pnl": 15.0, + "paper_avg_latency_ms": 19.0, + "live_trade_count": 2, + "live_total_pnl": 27.3, + "live_avg_pnl": 13.65, + "live_avg_latency_ms": 29.0, + "pnl_diff_mean": -1.3499999999999996, + "pnl_diff_std": 3.8499999999999996 +} \ No newline at end of file diff --git a/QuantTrader/scripts/live_trade.py b/QuantTrader/scripts/live_trade.py new file mode 100644 index 0000000..fd7a967 --- /dev/null +++ b/QuantTrader/scripts/live_trade.py @@ -0,0 +1,58 @@ +""" +简易实盘脚本示例（使用 OANDA） +将实时数据传入策略，策略产生信号 -> 风险检查 -> 下单执行 + +注意：在真实交易前请先在沙盒/模拟账户充分测试。 +""" +import asyncio +import os +from dotenv import load_dotenv +from loguru import logger + +from core.data.oanda import OANDADataFeed +from core.execution.oanda_handler import OANDAExecutionHandler +from core.strategy.rsi_mean_reversion import RSIMeanReversionStrategy +from core.risk.base import SimpleRiskManager +from core.data.base import MarketDataEvent + +load_dotenv() + +async def main(): + account_id = os.getenv('OANDA_ACCOUNT_ID') + token = os.getenv('OANDA_TOKEN') + env = os.getenv('OANDA_ENVIRONMENT', 'practice') + instrument = 'EUR_USD' + + data_feed = OANDADataFeed( + instrument=instrument, + timeframe='H1', + account_id=account_id, + access_token=token, + environment=env + ) + + strategy = RSIMeanReversionStrategy(instrument=instrument) + risk_manager = SimpleRiskManager(max_position_size=1.0, max_portfolio_risk=0.02, max_drawdown=0.1) + execution = OANDAExecutionHandler(account_id=account_id, access_token=token, environment=env) + + async def on_market(event: MarketDataEvent): + signal = await strategy.on_data(event) + if not signal: + return + if await risk_manager.check_signal(signal): + await execution.process_signal(signal) + + # 订阅实时数据并运行 + await data_feed.subscribe(on_market) + logger.info('已订阅实时数据，开始监听（按 Ctrl+C 退出）') + + try: + while True: + await asyncio.sleep(1) + except asyncio.CancelledError: + pass + finally: + await data_feed.unsubscribe() + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/QuantTrader/scripts/paper_trade.py b/QuantTrader/scripts/paper_trade.py new file mode 100644 index 0000000..06d1b4c --- /dev/null +++ b/QuantTrader/scripts/paper_trade.py @@ -0,0 +1,296 @@ +""" +Paper trading driver that wires live OANDA pricing into the strategy engine. +""" + +from __future__ import annotations + +import argparse +import signal +import time +from queue import Empty, Queue + +import pandas as pd +from loguru import logger +import yaml + +import os +import sys + +TRADER_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +REPO_ROOT = os.path.dirname(TRADER_ROOT) +RESEARCH_ROOT = os.path.join(REPO_ROOT, "QuantResearch") +sys.path.extend([TRADER_ROOT, REPO_ROOT, RESEARCH_ROOT]) + +from data.oanda_stream import OandaPricingStream +from core.oanda_execution import OandaExecution +from core.events import TickEvent, OrderEvent +from QuantResearch.core.backtest.strategy_engine import ( + StrategyEngine, + StrategySpec, + parse_strategy_specs, + _coerce_fx_rates, + _merge_fx_rates, +) +from shared.utils.config import OANDA_ACCOUNT_ID, OANDA_TOKEN + + +class BarAggregator: + """ + Aggregate tick data into fixed timeframe OHLC bars. + """ + + def __init__(self, symbol: str, timeframe: str): + self.symbol = symbol.replace("_", "").upper() + tf = timeframe.lower() if isinstance(timeframe, str) else timeframe + self.timeframe = pd.to_timedelta(tf) + if self.timeframe <= pd.Timedelta(0): + raise ValueError(f"Invalid timeframe: {timeframe}") + self.current_bucket: pd.Timestamp | None = None + self.open = self.high = self.low = self.close = None + self.last_ts: pd.Timestamp | None = None + + def update(self, tick: TickEvent) -> dict | None: + ts = pd.Timestamp(tick.ts) + bucket = ts.floor(self.timeframe) + mid = (tick.bid + tick.ask) / 2.0 + if self.current_bucket is None: + self._start_bar(bucket, mid, ts) + return None + if bucket != self.current_bucket: + finished = self._build_bar() + self._start_bar(bucket, mid, ts) + return finished + self._update_bar(mid, ts) + return None + + def flush(self) -> dict | None: + if self.current_bucket is None: + return None + return self._build_bar() + + def _start_bar(self, bucket: pd.Timestamp, price: float, ts: pd.Timestamp) -> None: + self.current_bucket = bucket + self.open = self.high = self.low = self.close = price + self.last_ts = ts + + def _update_bar(self, price: float, ts: pd.Timestamp) -> None: + self.high = max(self.high, price) + self.low = min(self.low, price) + self.close = price + self.last_ts = ts + + def _build_bar(self) -> dict: + bar = { + "symbol": self.symbol, + "ts": self.last_ts, + "open": float(self.open), + "high": float(self.high), + "low": float(self.low), + "close": float(self.close), + "volume": 0, + } + return bar + + +def load_config(path: str) -> dict: + with open(path, "r", encoding="utf-8") as f: + return yaml.safe_load(f) or {} + + +def build_engine(cfg: dict, args, execution_handler): + symbol = args.symbol or cfg.get("symbol", "EURUSD") + account_ccy = cfg.get("account_ccy", "USD") + fast_win = int(cfg.get("fast", 50)) + slow_win = int(cfg.get("slow", 200)) + spread = float(cfg.get("spread", 1.0)) + slip = float(cfg.get("slip", 0.2)) + comm = float(cfg.get("comm", 2.0)) + qty = float(cfg.get("qty", 10_000)) + initial_cash = float(cfg.get("cash", 100_000)) + stop_loss_pips = cfg.get("sl", 50) + take_profit_pips = cfg.get("tp") + atr_sl = cfg.get("atr_sl") + atr_tp = cfg.get("atr_tp") + atr_window = int(cfg.get("atr_window", 14)) + regime_ema_window = int(cfg.get("regime_ema_window", 200)) + regime_slope_min = cfg.get("regime_slope_min") + if regime_slope_min is not None: + regime_slope_min = float(regime_slope_min) + regime_atr_min = cfg.get("regime_atr_min") + if regime_atr_min is not None: + regime_atr_min = float(regime_atr_min) + rsi_period = int(cfg.get("rsi_period", 14)) + rsi_long_thresh = cfg.get("rsi_long_thresh") + if rsi_long_thresh is not None: + rsi_long_thresh = float(rsi_long_thresh) + rsi_short_thresh = cfg.get("rsi_short_thresh") + if rsi_short_thresh is not None: + rsi_short_thresh = float(rsi_short_thresh) + enable_trailing = bool(cfg.get("enable_trailing", False)) + trailing_enable_atr_mult = float(cfg.get("trailing_enable_atr_mult", 1.0)) + trailing_atr_mult = float(cfg.get("trailing_atr_mult", 0.5)) + long_only_above_slow = bool(cfg.get("long_only_above_slow", False)) + slope_lookback = int(cfg.get("slope_lookback", 0)) + cooldown = int(cfg.get("cooldown", 0)) + allow_short = bool(cfg.get("allow_short", True)) + short_only_below_slow = bool(cfg.get("short_only_below_slow", False)) + risk_per_trade_pct = cfg.get("risk_per_trade_pct") + max_drawdown_pct = cfg.get("max_drawdown_pct") + max_position_units = cfg.get("max_position_units") + htf_factor = int(cfg.get("htf_factor", 4)) + htf_ema_window = cfg.get("htf_ema_window") + if htf_ema_window is not None: + htf_ema_window = int(htf_ema_window) + htf_rsi_period = cfg.get("htf_rsi_period") + if htf_rsi_period is not None: + htf_rsi_period = int(htf_rsi_period) + + cfg_fx_rates = _coerce_fx_rates(cfg.get("fx_rates")) + cli_fx_rates = _coerce_fx_rates(args.fx_rate) + fx_rates = _merge_fx_rates(cfg_fx_rates, cli_fx_rates) + + strategy_specs = parse_strategy_specs(cfg.get("strategies")) + + engine = StrategyEngine( + symbol=symbol, + fast_win=fast_win, + slow_win=slow_win, + spread_pips=spread, + commission_per_million=comm, + slippage_pips=slip, + stop_loss_pips=stop_loss_pips, + take_profit_pips=take_profit_pips, + atr_sl=atr_sl, + atr_tp=atr_tp, + atr_window=atr_window, + regime_ema_window=regime_ema_window, + regime_slope_min=regime_slope_min, + regime_atr_min=regime_atr_min, + rsi_period=rsi_period, + rsi_long_thresh=rsi_long_thresh, + rsi_short_thresh=rsi_short_thresh, + enable_trailing=enable_trailing, + trailing_enable_atr_mult=trailing_enable_atr_mult, + trailing_atr_mult=trailing_atr_mult, + long_only_above_slow=long_only_above_slow, + slope_lookback=slope_lookback, + cooldown=cooldown, + qty=qty, + account_ccy=account_ccy, + fx_rates=fx_rates, + strategy_specs=strategy_specs, + allow_short=allow_short, + short_only_below_slow=short_only_below_slow, + risk_per_trade_pct=risk_per_trade_pct, + max_drawdown_pct=max_drawdown_pct, + max_position_units=max_position_units, + htf_factor=htf_factor, + htf_ema_window=htf_ema_window, + htf_rsi_period=htf_rsi_period, + execution_handler=execution_handler, + ) + engine.set_initial_cash(initial_cash) + return engine, symbol, initial_cash + + +def main(): + parser = argparse.ArgumentParser(description="OANDA paper trading driver") + parser.add_argument("--config", required=True, help="策略配置 YAML") + parser.add_argument("--symbol", default=None, help="覆盖配置中的交易品种") + parser.add_argument("--timeframe", default="60s", help="K线时间粒度，默认 60s") + parser.add_argument("--environment", default="practice", choices=["practice", "live"], help="OANDA 环境") + parser.add_argument("--fx-rate", action="append", default=None, help="额外汇率，示例 GBPUSD=1.27") + parser.add_argument("--max-bars", type=int, default=None, help="最多生成多少根 bar 后自动停止") + parser.add_argument("--log-heartbeat", action="store_true", help="打印 OANDA 心跳信息") + args = parser.parse_args() + + cfg = load_config(args.config) + token = OANDA_TOKEN + account_id = OANDA_ACCOUNT_ID + if not token or not account_id: + raise RuntimeError("OANDA_TOKEN 或 OANDA_ACCOUNT_ID 未在环境变量中设置") + + order_queue: Queue = Queue() + execution = OandaExecution(order_queue, account_id=account_id, access_token=token, environment=args.environment) + engine, symbol, initial_cash = build_engine(cfg, args, execution.on_event) + aggregator = BarAggregator(symbol, args.timeframe) + + tick_queue: Queue = Queue() + stream = OandaPricingStream( + tick_queue, + account_id=account_id, + instruments=[symbol], + access_token=token, + environment=args.environment, + log_heartbeat=args.log_heartbeat, + ) + + stop_flag = False + + def handle_sigterm(signum, frame): + nonlocal stop_flag + stop_flag = True + + signal.signal(signal.SIGINT, handle_sigterm) + signal.signal(signal.SIGTERM, handle_sigterm) + + logger.info(f"[Paper] Starting pricing stream for {symbol} ({args.timeframe})") + stream.start() + + bars_processed = 0 + try: + while not stop_flag: + try: + tick = tick_queue.get(timeout=1.0) + except Empty: + continue + if not isinstance(tick, TickEvent): + continue + bar = aggregator.update(tick) + if bar: + engine.handle_bar(bar) + bars_processed += 1 + if args.max_bars and bars_processed >= args.max_bars: + logger.info("[Paper] Reached max bar limit, stopping.") + break + finally: + stream.stop() + + # flush last partially built bar + final_bar = aggregator.flush() + if final_bar: + engine.handle_bar(final_bar) + + engine.finalize() + suffix = engine.compute_suffix() + engine.export_outputs( + fast_win=int(cfg.get("fast", 50)), + slow_win=int(cfg.get("slow", 200)), + suffix=suffix, + ) + result = engine.summary( + fast_win=int(cfg.get("fast", 50)), + slow_win=int(cfg.get("slow", 200)), + suffix=suffix, + ) + final_equity = result["final_equity"] if result["final_equity"] is not None else engine.cash + ret_pct = (final_equity / initial_cash - 1.0) * 100.0 + logger.info(f"[Paper] Bars processed: {engine.bar_count}, Trades executed: {engine.trade_count}") + logger.info(f"[Paper] Final equity: {final_equity:.2f} ({ret_pct:.2f}%)") + + # Drain any fills left in queue + fills = [] + while True: + try: + fill = order_queue.get_nowait() + except Empty: + break + else: + fills.append(fill) + if fills: + for fill in fills: + logger.info(f"[Paper] Fill received: {fill}") + + +if __name__ == "__main__": + main() diff --git a/QuantTrader/tests/__init__.py b/QuantTrader/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/QuantTrader/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/QuantTrader/tests/__pycache__/__init__.cpython-312.pyc b/QuantTrader/tests/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..a640ccf Binary files /dev/null and b/QuantTrader/tests/__pycache__/__init__.cpython-312.pyc differ diff --git a/QuantTrader/tests/__pycache__/test_execution_adapters.cpython-312.pyc b/QuantTrader/tests/__pycache__/test_execution_adapters.cpython-312.pyc new file mode 100644 index 0000000..8a031b8 Binary files /dev/null and b/QuantTrader/tests/__pycache__/test_execution_adapters.cpython-312.pyc differ diff --git a/QuantTrader/tests/__pycache__/test_risk_engine.cpython-312.pyc b/QuantTrader/tests/__pycache__/test_risk_engine.cpython-312.pyc new file mode 100644 index 0000000..6059438 Binary files /dev/null and b/QuantTrader/tests/__pycache__/test_risk_engine.cpython-312.pyc differ diff --git a/QuantTrader/tests/test_execution_adapters.py b/QuantTrader/tests/test_execution_adapters.py new file mode 100644 index 0000000..544c75e --- /dev/null +++ b/QuantTrader/tests/test_execution_adapters.py @@ -0,0 +1,63 @@ +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock + +from QuantTrader.execution.adapter import OrderParams +from QuantTrader.execution.oanda_adapter import OandaAdapter +from QuantTrader.execution.order_store import OrderStore +from QuantTrader.execution.paper_adapter import PaperAdapter + + +class ExecutionAdaptersTest(unittest.TestCase): + def setUp(self): + self.tmpdir = tempfile.TemporaryDirectory() + os.environ["EXECUTION_METRICS_PATH"] = str( + (tempfile.NamedTemporaryFile(delete=False, dir=self.tmpdir.name).name) + ) + + def tearDown(self): + self.tmpdir.cleanup() + os.environ.pop("EXECUTION_METRICS_PATH", None) + + def test_oanda_adapter_submit_uses_order_store(self): + store_path = f"{self.tmpdir.name}/orders.log" + adapter = OandaAdapter( + account_id="ACC", + token="TOKEN", + order_store=OrderStore(store_path), + base_url="https://example.com", + metrics_path=os.environ["EXECUTION_METRICS_PATH"], + ) + stub_response = { + "orderCreateTransaction": { + "id": "123", + "time": "2024-01-01T00:00:00.000000Z", + } + } + adapter._request = MagicMock(return_value=stub_response) # type: ignore + order = OrderParams(symbol="EUR_USD", side="buy", quantity=1000) + ack = adapter.submit(order) + self.assertEqual(ack.order_id, "123") + self.assertTrue(Path(store_path).exists()) + + def test_paper_adapter_generates_ids_and_logs_equity(self): + store_path = f"{self.tmpdir.name}/paper.log" + equity_path = f"{self.tmpdir.name}/equity.csv" + adapter = PaperAdapter( + latency_ms=1, + slippage_pips=0.0, + order_store=OrderStore(store_path), + equity_log_path=equity_path, + ) + order = OrderParams(symbol="EURUSD", side="buy", quantity=1000, price=1.1) + ack = adapter.submit(order) + self.assertTrue(ack.order_id.startswith("PAPER-")) + cancel_ack = adapter.cancel(ack.order_id) + self.assertEqual(cancel_ack.status, "cancelled") + self.assertTrue(Path(equity_path).exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/QuantTrader/tests/test_risk_engine.py b/QuantTrader/tests/test_risk_engine.py new file mode 100644 index 0000000..7b30eb4 --- /dev/null +++ b/QuantTrader/tests/test_risk_engine.py @@ -0,0 +1,34 @@ +import unittest + +from QuantTrader.core.risk.risk_engine import RiskEngine, RiskLimits + + +class RiskEngineTest(unittest.TestCase): + def setUp(self): + limits = RiskLimits( + max_position_notional=100000, + max_gross_leverage=2.0, + max_daily_loss=5000, + max_drawdown=0.1, + ) + self.engine = RiskEngine(limits=limits, starting_equity=50000) + + def test_exposure_limit(self): + ok, _ = self.engine.evaluate_order("EURUSD", "buy", 90000) + self.assertTrue(ok) + self.engine.record_fill("EURUSD", "buy", 90000, pnl=0) + ok, reason = self.engine.evaluate_order("EURUSD", "buy", 20000) + self.assertFalse(ok) + self.assertEqual(reason, "symbol_exposure_limit:EURUSD") + + def test_daily_loss_limit(self): + ok, _ = self.engine.check_loss_limits() + self.assertTrue(ok) + self.engine.record_fill("USDJPY", "sell", 50000, pnl=-6000) + ok, reason = self.engine.check_loss_limits() + self.assertFalse(ok) + self.assertEqual(reason, "daily_loss_limit") + + +if __name__ == "__main__": + unittest.main() diff --git a/README.md b/README.md new file mode 100644 index 0000000..9cb7c9b --- /dev/null +++ b/README.md @@ -0,0 +1,168 @@ +# FX Backtest & Execution Stack + +End-to-end foreign-exchange research and trading toolkit that links feature engineering, backtesting, ML-driven signal generation, and OANDA execution into one repo. + +## Highlights + +- Unified `StrategyEngine` (see `QuantResearch/core/backtest/strategy_engine.py`) powers historical backtests, walk-forward studies, paper trading, and the live runner so signals behave identically across environments. +- Strategy registry ships with SMA/ATR trend, Bollinger & band mean-revert, breakout momentum, and an XGBoost probability model (`QuantResearch/strategies/*`), allowing multi-strategy voting through YAML configs such as `QuantTrader/config/usdjpy_multi_strategy.yaml`. +- Research workflows enforce data-manifest validation, risk sims, KPI summaries (`results//summary.json`), and promotion of vetted artifacts into `QuantTrader/artifacts/` before they are allowed to reach trading. +- Runtime layer contains async OANDA data/execution handlers, event-driven risk checks, and pluggable multi-strategy allocation for both paper (`scripts/paper_trade.py`) and live trading (`scripts/live_trade.py`). +- Monitoring stack (Pushgateway + Prometheus + Grafana) ships ready-to-import risk dashboards (`monitoring/grafana/*.json`), custom drilldown plugins (logs/traces/profiles/metrics), and Slack/pushgateway hooks for diagnostics automation. + +## Repository Layout + +- `QuantResearch/` – Research code, datasets, strategy implementations, notebooks/scripts, docs, artifacts, and test suites. +- `QuantTrader/` – Trading runtime with execution/risk/data engines, configs, logging, and artifact promotion targets. +- `monitoring/` – Dockerized observability stack plus Grafana dashboards & plugins for metrics/logs/traces/profiles. +- `shared/` – Cross-cutting helpers (`shared/utils/config.py` loads OANDA/Slack/Pushgateway secrets from `.env`). +- `results/` – Canonical run outputs uploaded with PRs (e.g., walk-forward summaries) for auditing. +- `metrics/` – Lightweight operational CSVs (e.g., execution latencies) that can be pushed to Prometheus. + +## Quick Start + +1. **Clone & create a virtual environment** + + ```bash + git clone + cd FX_Backtest + python -m venv .venv + source .venv/bin/activate + pip install --upgrade pip + pip install -r QuantResearch/requirements.txt + pip install -r QuantTrader/requirements.txt + ``` + + Python 3.10+ is recommended for `pandas`/`xgboost` compatibility. + +2. **Configure secrets** + + ```bash + cp .env.demo .env + # edit .env with your OANDA practice/live credentials + webhook URLs + source .env + ``` + + All scripts that touch OANDA import from `shared.utils.config`, so missing env vars fail fast. + +3. **Prepare data** + + - Drop raw CSVs (e.g., `USDJPY_H1.csv`) under `QuantResearch/data/raw/`. + - Rebuild the manifest + integrity reports any time data changes: + + ```bash + cd QuantResearch + python scripts/build_dataset_manifest.py --dirs data/raw data/derived --output data/_manifest.json + python scripts/check_data_integrity.py + ``` + +4. **Run a backtest** + + ```bash + python QuantResearch/scripts/backtest_strategy.py \ + --csv QuantResearch/data/raw/USDJPY_H1.csv \ + --symbol USDJPY \ + --fast 20 --slow 80 \ + --strategies QuantTrader/config/usdjpy_multi_strategy.yaml + ``` + + The script validates the dataset, runs the engine, and writes KPIs plus `equity/`, `trades/`, and `stats/` artifacts under `QuantResearch/data/outputs/`. + +5. **Train or refresh the XGBoost signal** + + ```bash + python QuantResearch/scripts/train_xgb_usdjpy.py \ + --csv QuantResearch/data/raw/USDJPY_H1.csv \ + --symbol USDJPY \ + --out QuantResearch/artifacts/models/usdjpy_h1_xgb + ``` + + This exports `model.json`, feature lists, thresholds, and updates `usdjpy_h1_xgb_latest.json` so trading configs can point to the latest model. + +6. **Run walk-forward analysis (optional gating)** + + ```bash + python QuantResearch/scripts/run_walkforward.py \ + --config QuantTrader/config/usdjpy_multi_strategy.yaml \ + --csv QuantResearch/data/raw/USDJPY_H1.csv \ + --train-bars 4000 --test-bars 1000 \ + --output-root QuantResearch/results \ + --label usdjpy_xgb + ``` + + Each window produces metrics and a `summary.json` under `QuantResearch/results//`. Reference these run IDs in PRs. + +7. **Promote artifacts to the trader** + + After validating a run, sync configs/params into `QuantTrader/artifacts/` (see `QuantTrader/artifacts/README.md`): + + ```bash + cp QuantTrader/config/usdjpy_multi_strategy.yaml QuantTrader/artifacts/config/ + cp QuantResearch/artifacts/models/usdjpy_h1_xgb_latest.json QuantTrader/artifacts/params/ + ``` + +8. **Paper trading or live execution** + + - Paper (uses live pricing -> StrategyEngine -> simulated fills): + + ```bash + python QuantTrader/scripts/paper_trade.py \ + --config QuantTrader/config/usdjpy_multi_strategy.yaml \ + --symbol USDJPY \ + --timeframe 60s + ``` + + - Live example (direct OANDA handler + RSI strategy template, see `QuantTrader/scripts/live_trade.py`): + + ```bash + python QuantTrader/scripts/live_trade.py + ``` + + Customize the risk manager, strategy, and execution handler before pointing to a funded account. + +9. **Spin up monitoring (optional but recommended)** + + ```bash + docker compose up -d + ``` + + This launches Pushgateway (`:9091`), Prometheus (`:9090`), and Grafana (`:3000`). Import `monitoring/grafana/risk_metrics_dashboard.json` and enable the bundled drilldown plugins for logs/traces/profiles/metrics exploration. + +## Common Workflows + +- **Data quality gating:** `python QuantResearch/scripts/watch_quality.py` or the CI-friendly `scripts/watch_risk_metrics.py` push metrics to Slack/Pushgateway before PRs merge. +- **Batch experiments:** `python QuantResearch/scripts/run_batch_backtests.py --config config/eurusd_grid.yaml` sweeps parameter grids and streams metrics under `results//`. +- **Stress testing:** `python QuantResearch/scripts/validate_stress_scenarios.py --config ...` replays adverse cost scenarios to validate drawdown budgets. +- **Risk sims:** `RUN= ./QuantResearch/scripts/run_risk_sim.sh && ./QuantResearch/bin/backfill_risk.sh` keep `results/risk/metrics.csv` aligned with latest runs. + +## Monitoring & Diagnostics + +- `QuantResearch/scripts/export_metrics_prom.py` streams aggregated KPIs to Pushgateway (`PUSHGATEWAY_URL`). +- `QuantResearch/scripts/notify_risk_metrics.sh` wraps `watch_risk_metrics.py` to send Slack alerts using `SLACK_RISK_WEBHOOK`. +- Grafana plugins under `monitoring/grafana/plugins/grafana-*-app/` document the queryless drilldown experiences for logs (Loki), metrics (Prometheus), traces (Tempo), and profiles (Pyroscope). +- `monitoring/grafana/risk_metrics_dashboard.json` visualizes walk-forward pass rates, tail risk, exposure, and per-strategy attribution. Load it after Grafana boots (`admin/admin` by default). + +## Testing & Validation + +- Unit tests: `pytest QuantResearch/tests QuantTrader/tests`. +- Strategy registry coverage: `QuantResearch/tests/test_strategy_registry.py` ensures new strategies register correctly; add fixtures before contributing. +- Result validation: `python QuantResearch/scripts/validate_results.py QuantResearch/results/` checks KPI completeness + data references. +- Data feed/execution smoke tests: `python QuantTrader/tests/test_execution_adapters.py` mocks OANDA flows. + +## Extending the Stack + +1. Implement a new research strategy under `QuantResearch/strategies/` and decorate it with `@register("my_strategy")`. +2. Reference it inside a config YAML (e.g., `usdjpy_multi_strategy.yaml`) with weights/params. +3. Add risk rules in `QuantTrader/core/risk/` if the position sizing model needs to change. +4. Document any new process in `QuantResearch/docs/` or module-level READMEs so CI reviewers have breadcrumbs. + +## Related Docs + +- `QuantResearch/README.md` – data submission rules, risk/diagnostics workflow. +- `QuantTrader/artifacts/README.md` – promotion checklist for configs/params. +- `monitoring/grafana/plugins/*/README.md` – upstream plugin instructions. + +## License + +No open-source license is declared yet. Keep the repository private or add a LICENSE file before publishing. + diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..1e2b722 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,32 @@ +version: "3.8" +services: + pushgateway: + image: prom/pushgateway:latest + container_name: pushgateway + ports: + - "9091:9091" + restart: unless-stopped + + prometheus: + image: prom/prometheus:latest + container_name: prometheus + ports: + - "9090:9090" + volumes: + - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro + depends_on: + - pushgateway + restart: unless-stopped + + grafana: + image: grafana/grafana:latest + container_name: grafana + ports: + - "3000:3000" + volumes: + - ./monitoring/grafana:/var/lib/grafana + environment: + - GF_SECURITY_ADMIN_PASSWORD=admin + depends_on: + - prometheus + restart: unless-stopped diff --git a/main.py b/main.py new file mode 100644 index 0000000..d8a956b --- /dev/null +++ b/main.py @@ -0,0 +1,23 @@ +# main.py +import oandapyV20 +import oandapyV20.endpoints.accounts as accounts +from shared.utils.config import OANDA_TOKEN, OANDA_ACCOUNT_ID, OANDA_URL +from shared.utils.logger import logger + +def connect_oanda(): + client = oandapyV20.API(access_token=OANDA_TOKEN) + return client + +def get_account_summary(client): + r = accounts.AccountSummary(accountID=OANDA_ACCOUNT_ID) + client.request(r) + return r.response + +if __name__ == "__main__": + client = connect_oanda() + logger.info("✅ Connected to OANDA") + + summary = get_account_summary(client) + balance = summary['account']['balance'] + currency = summary['account']['currency'] + logger.info(f"💰 Balance: {balance} {currency}") diff --git a/metrics/execution.csv b/metrics/execution.csv new file mode 100644 index 0000000..3093553 --- /dev/null +++ b/metrics/execution.csv @@ -0,0 +1,5 @@ +event,order_id,symbol,latency_ms,status,timestamp +submit,PAPER-1,USDJPY,54.391000000000005,accepted,2025-11-11T13:38:55.591742 +submit,PAPER-1,USDJPY,26.575000000000003,accepted,2025-11-11T13:39:09.656374 +submit,PAPER-1,USDJPY,30.724,accepted,2025-11-11T13:39:26.674012 +submit,PAPER-1,USDJPY,26.801,accepted,2025-11-11T13:39:38.077863