yuqei
140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
from pathlib import Path
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Optional
|
|
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.services.backtest import (
|
|
BacktestCostConfig,
|
|
BacktestQualityGate,
|
|
BacktestService,
|
|
evaluate_backtest_quality,
|
|
summarize_backtest_outcomes,
|
|
)
|
|
|
|
|
|
def build_cost_config_from_env(env=None) -> BacktestCostConfig:
|
|
env = os.environ if env is None else env
|
|
return BacktestCostConfig(
|
|
fee_rate=_get_float(env, "BACKTEST_FEE_RATE", 0.0),
|
|
slippage_rate=_get_float(env, "BACKTEST_SLIPPAGE_RATE", 0.0),
|
|
fixed_cost_r=_get_float(env, "BACKTEST_FIXED_COST_R", 0.0),
|
|
)
|
|
|
|
|
|
def build_require_quality_gate_from_env(env=None) -> bool:
|
|
env = os.environ if env is None else env
|
|
value = str(env.get("BACKTEST_REQUIRE_QUALITY_GATE") or "").strip().lower()
|
|
return value in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def build_quality_gate_from_env(env=None) -> BacktestQualityGate:
|
|
env = os.environ if env is None else env
|
|
defaults = BacktestQualityGate()
|
|
return BacktestQualityGate(
|
|
min_total_signals=_get_int(env, "BACKTEST_MIN_TOTAL_SIGNALS", defaults.min_total_signals),
|
|
min_executed_signals=_get_int(env, "BACKTEST_MIN_EXECUTED_SIGNALS", defaults.min_executed_signals),
|
|
min_expectancy_r=_get_float(env, "BACKTEST_MIN_EXPECTANCY_R", defaults.min_expectancy_r),
|
|
min_profit_factor_r=_get_float(env, "BACKTEST_MIN_PROFIT_FACTOR_R", defaults.min_profit_factor_r),
|
|
max_drawdown_r=_get_float(env, "BACKTEST_MAX_DRAWDOWN_R", defaults.max_drawdown_r),
|
|
max_consecutive_losses=_get_int(env, "BACKTEST_MAX_CONSECUTIVE_LOSSES", defaults.max_consecutive_losses),
|
|
)
|
|
|
|
|
|
def run_backtest_once(
|
|
service: BacktestService,
|
|
instrument_id: int = 1,
|
|
limit: int = 100,
|
|
cost_config: Optional[BacktestCostConfig] = None,
|
|
quality_gate: Optional[BacktestQualityGate] = None,
|
|
require_quality_gate: bool = False,
|
|
output_json: bool = False,
|
|
) -> int:
|
|
run_id, outcomes = service.run_backtest(
|
|
instrument_id=instrument_id,
|
|
limit=limit,
|
|
cost_config=cost_config,
|
|
quality_gate=quality_gate,
|
|
)
|
|
summary = summarize_backtest_outcomes(outcomes)
|
|
quality_gate_result = evaluate_backtest_quality(summary, gate=quality_gate)
|
|
payload = {
|
|
"passed": not require_quality_gate or quality_gate_result["passed"],
|
|
"backtest_run_id": run_id,
|
|
"result_count": len(outcomes),
|
|
"summary": summary,
|
|
"quality_gate": quality_gate_result,
|
|
"quality_gate_required": require_quality_gate,
|
|
}
|
|
exit_code = 2 if require_quality_gate and not quality_gate_result["passed"] else 0
|
|
|
|
if output_json:
|
|
print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
|
|
else:
|
|
print(f"Backtest run id: {run_id}; results written: {len(outcomes)}")
|
|
print(f"Backtest quality gate passed: {quality_gate_result['passed']}")
|
|
if quality_gate_result["reasons"]:
|
|
print(f"Backtest quality gate reasons: {', '.join(quality_gate_result['reasons'])}")
|
|
return exit_code
|
|
|
|
|
|
def main(env=None) -> int:
|
|
env = os.environ if env is None else env
|
|
output_json = _get_bool(env, "BACKTEST_OUTPUT_JSON", False)
|
|
try:
|
|
service = BacktestService()
|
|
cost_config = build_cost_config_from_env(env=env)
|
|
return run_backtest_once(
|
|
service=service,
|
|
cost_config=cost_config,
|
|
quality_gate=build_quality_gate_from_env(env=env),
|
|
require_quality_gate=build_require_quality_gate_from_env(env=env),
|
|
output_json=output_json,
|
|
)
|
|
except SQLAlchemyError as exc:
|
|
summary = {
|
|
"passed": False,
|
|
"backtest_run_id": None,
|
|
"result_count": 0,
|
|
"summary": None,
|
|
"quality_gate": None,
|
|
"error_type": exc.__class__.__name__,
|
|
"error": str(exc),
|
|
}
|
|
if output_json:
|
|
print(json.dumps(summary, default=str, ensure_ascii=False, sort_keys=True))
|
|
else:
|
|
print(f"Database unavailable for backtest sample: {exc.__class__.__name__}")
|
|
return 2
|
|
|
|
|
|
def _get_float(env, key: str, default: float) -> float:
|
|
value = env.get(key)
|
|
if value is None or value == "":
|
|
return default
|
|
return float(value)
|
|
|
|
|
|
def _get_int(env, key: str, default: int) -> int:
|
|
value = env.get(key)
|
|
if value is None or value == "":
|
|
return default
|
|
return int(value)
|
|
|
|
|
|
def _get_bool(env, key: str, default: bool) -> bool:
|
|
value = env.get(key)
|
|
if value is None or value == "":
|
|
return default
|
|
return str(value).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|