rckrdmrd/trading-platform-ml-engine-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[GAP-3] feat: Wire RRBacktester into prediction service and API

Browse Source 
- Added RRBacktester import and initialization in PredictionService
- Added BacktestSummary dataclass for typed responses
- Added run_backtest() method with signal generation
- Added _generate_backtest_signals() helper for historical signals
- Added backtester_available property
- Updated POST /api/backtest endpoint to use real backtester
- Added GET /api/backtest/status endpoint for availability check
- Added losing_trades, avg_trade_duration, metrics_by_rr, metrics_by_amd to response

The backtester now uses:
- RRBacktester from src/backtesting/rr_backtester.py
- MetricsCalculator from src/backtesting/metrics.py
- AMD phase filtering and R:R configuration support

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Adrian Flores Cortes 2026-01-25 07:13:26 -06:00
parent a22fb11968
commit a1e606c21a
2 changed files with 290 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
src/api/main.py
View File

@ -25,7 +25,8 @@ from ..services.prediction_service import (
Direction,
AMDPhase as ServiceAMDPhase,
VolatilityRegime as ServiceVolatilityRegime,
HierarchicalResult
HierarchicalResult,
BacktestSummary
)
# API Models
@ -833,6 +834,7 @@ class BacktestResponse(BaseModel):
"""Backtest results response"""
total_trades: int
winning_trades: int
losing_trades: int
winrate: float
net_profit: float
profit_factor: float
@ -843,41 +845,105 @@ class BacktestResponse(BaseModel):
signals_generated: int
signals_filtered: int
signals_traded: int
avg_trade_duration: float = 0.0
metrics_by_rr: Optional[Dict[str, Any]] = None
metrics_by_amd: Optional[Dict[str, Any]] = None
@app.post("/api/backtest", response_model=BacktestResponse, tags=["Backtesting"])
async def run_backtest(request: BacktestRequest):
"""
Run backtest on historical data
Run backtest on historical data.
Simulates trading based on ML predictions and returns performance metrics.
Uses RRBacktester with TP/SL simulation and AMD phase filtering.
Args:
request: Backtest configuration
request: Backtest configuration including:
- symbol: Trading symbol
- start_date/end_date: Date range for backtest
- initial_capital: Starting capital
- risk_per_trade: Risk per trade as fraction
- rr_config: Risk/Reward config (rr_2_1 or rr_3_1)
- min_confidence: Minimum probability to trade
- filter_by_amd: Whether to filter by AMD phase
"""
if not models_state.get("backtester"):
global prediction_service
if prediction_service is None:
prediction_service = get_prediction_service()
if not prediction_service.backtester_available:
raise HTTPException(
status_code=503,
detail="Backtester not loaded"
detail="Backtester not available. Service may still be initializing."
)
# TODO: Implement actual backtesting
# backtester = models_state["backtester"]
# result = backtester.run_backtest(price_data, signals)
try:
result = await prediction_service.run_backtest(
symbol=request.symbol.upper(),
start_date=request.start_date,
end_date=request.end_date,
initial_capital=request.initial_capital,
risk_per_trade=request.risk_per_trade,
rr_config=request.rr_config,
min_confidence=request.min_confidence,
filter_by_amd=request.filter_by_amd
)
# Mock response
return BacktestResponse(
total_trades=150,
winning_trades=82,
winrate=0.547,
net_profit=3250.75,
profit_factor=1.85,
max_drawdown=1250.50,
max_drawdown_pct=0.125,
sharpe_ratio=1.42,
sortino_ratio=2.15,
signals_generated=450,
signals_filtered=200,
signals_traded=150
)
if result is None:
raise HTTPException(
status_code=500,
detail="Backtest failed. Check logs for details."
)
return BacktestResponse(
total_trades=result.total_trades,
winning_trades=result.winning_trades,
losing_trades=result.losing_trades,
winrate=result.winrate,
net_profit=result.net_profit,
profit_factor=result.profit_factor,
max_drawdown=result.max_drawdown,
max_drawdown_pct=result.max_drawdown_pct,
sharpe_ratio=result.sharpe_ratio,
sortino_ratio=result.sortino_ratio,
signals_generated=result.signals_generated,
signals_filtered=result.signals_filtered,
signals_traded=result.signals_traded,
avg_trade_duration=result.avg_trade_duration,
metrics_by_rr=result.metrics_by_rr,
metrics_by_amd=result.metrics_by_amd
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Backtest failed: {e}")
raise HTTPException(status_code=500, detail=f"Backtest failed: {str(e)}")
@app.get("/api/backtest/status", tags=["Backtesting"])
async def get_backtest_status():
"""
Get backtester availability status.
"""
global prediction_service
if prediction_service is None:
prediction_service = get_prediction_service()
return {
"available": prediction_service.backtester_available,
"supported_symbols": ["XAUUSD", "EURUSD", "BTCUSD", "GBPUSD", "USDJPY"],
"rr_configs": ["rr_2_1", "rr_3_1"],
"default_config": {
"initial_capital": 10000.0,
"risk_per_trade": 0.02,
"min_confidence": 0.55,
"filter_by_amd": True
}
}
# Training endpoint

201
src/services/prediction_service.py
View File

@ -41,6 +41,10 @@ from ..pipelines.hierarchical_pipeline import (
PredictionResult as HierarchicalPredictionResult
)
# Backtesting engine
from ..backtesting.rr_backtester import RRBacktester, BacktestConfig, BacktestResult
from ..backtesting.metrics import TradingMetrics, TradeRecord
class Direction(Enum):
LONG = "long"
@ -118,6 +122,27 @@ class HierarchicalResult:
trade_quality: str # 'high', 'medium', 'low', 'skip'
@dataclass
class BacktestSummary:
"""Summary of backtest results"""
total_trades: int
winning_trades: int
losing_trades: int
winrate: float
net_profit: float
profit_factor: float
max_drawdown: float
max_drawdown_pct: float
sharpe_ratio: float
sortino_ratio: float
signals_generated: int
signals_filtered: int
signals_traded: int
avg_trade_duration: float
metrics_by_rr: Dict[str, Dict] = None
metrics_by_amd: Dict[str, Dict] = None
@dataclass
class TradingSignal:
"""Complete trading signal"""
@ -187,6 +212,7 @@ class PredictionService:
self._amd_detector = None
self._attention_provider = None # Level 0 attention models
self._hierarchical_pipeline = None # L0→L1→L2 pipeline
self._backtester = None # Backtesting engine
self._models_loaded = False
# Symbol-specific trainers (nuevos modelos por símbolo/timeframe)
@ -258,6 +284,14 @@ class PredictionService:
logger.warning(f"HierarchicalPipeline initialization failed: {e}")
self._hierarchical_pipeline = None
# Initialize Backtesting Engine
try:
self._backtester = RRBacktester(BacktestConfig())
logger.info("✅ RRBacktester initialized")
except Exception as e:
logger.warning(f"RRBacktester initialization failed: {e}")
self._backtester = None
self._models_loaded = True
# Cargar modelos por símbolo si el feature flag está activo
@ -715,6 +749,173 @@ class PredictionService:
return None
return self._hierarchical_pipeline.get_model_info(symbol)
@property
def backtester_available(self) -> bool:
"""Check if backtester is available."""
return self._backtester is not None
async def run_backtest(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
initial_capital: float = 10000.0,
risk_per_trade: float = 0.02,
rr_config: str = "rr_2_1",
min_confidence: float = 0.55,
filter_by_amd: bool = True
) -> Optional[BacktestSummary]:
"""
Run backtest on historical data.
Args:
symbol: Trading symbol
start_date: Start date for backtest
end_date: End date for backtest
initial_capital: Starting capital
risk_per_trade: Risk per trade as fraction (0.02 = 2%)
rr_config: Risk/Reward configuration (rr_2_1 or rr_3_1)
min_confidence: Minimum confidence to trade
filter_by_amd: Filter by AMD phase
Returns:
BacktestSummary with results, or None if failed
"""
if not self._backtester:
logger.warning("Backtester not available")
return None
try:
# Configure backtester
config = BacktestConfig(
initial_capital=initial_capital,
risk_per_trade=risk_per_trade,
min_confidence=min_confidence,
filter_by_amd=filter_by_amd
)
self._backtester.config = config
# Fetch historical data
days_diff = (end_date - start_date).days
periods = days_diff * 288 # Approximate 5-minute bars per day
df = await self.get_market_data(symbol, "5m", lookback_periods=min(periods, 10000))
if df.empty:
logger.warning(f"No data available for backtest: {symbol}")
return None
# Filter by date range
df = df[(df.index >= start_date) & (df.index <= end_date)]
if len(df) < 100:
logger.warning(f"Insufficient data for backtest: {len(df)} bars")
return None
# Generate signals for the period
signals = await self._generate_backtest_signals(df, symbol, rr_config)
# Run backtest
result = self._backtester.run_backtest(
price_data=df,
signals=signals,
rr_config={'name': rr_config, 'sl': 5.0, 'tp': 10.0 if rr_config == 'rr_2_1' else 15.0}
)
# Convert to summary
return BacktestSummary(
total_trades=result.metrics.total_trades,
winning_trades=result.metrics.winning_trades,
losing_trades=result.metrics.losing_trades,
winrate=result.metrics.winrate,
net_profit=result.metrics.net_profit,
profit_factor=result.metrics.profit_factor,
max_drawdown=result.metrics.max_drawdown,
max_drawdown_pct=result.metrics.max_drawdown_pct,
sharpe_ratio=result.metrics.sharpe_ratio,
sortino_ratio=result.metrics.sortino_ratio,
signals_generated=result.signals_generated,
signals_filtered=result.signals_filtered,
signals_traded=result.signals_traded,
avg_trade_duration=result.metrics.avg_trade_duration,
metrics_by_rr={k: v.to_dict() for k, v in result.metrics_by_rr.items()},
metrics_by_amd={k: v.to_dict() for k, v in result.metrics_by_amd.items()}
)
except Exception as e:
logger.error(f"Backtest failed for {symbol}: {e}")
return None
async def _generate_backtest_signals(
self,
df: pd.DataFrame,
symbol: str,
rr_config: str
) -> pd.DataFrame:
"""
Generate signals for backtesting from price data.
Uses the prediction models to generate prob_tp_first and direction.
"""
signals = pd.DataFrame(index=df.index)
# Generate signals at regular intervals (every 12 bars = 1 hour for 5m data)
signal_interval = 12
n = len(df)
signals['prob_tp_first'] = np.nan
signals['direction'] = 'long'
signals['horizon'] = '15m'
signals['rr_config'] = rr_config
signals['amd_phase'] = None
signals['volatility_regime'] = 'medium'
signals['confidence'] = np.nan
for i in range(100, n, signal_interval):
try:
# Get slice of data for prediction
df_slice = df.iloc[:i+1]
# Generate features
features = self.feature_engineer.create_features(df_slice)
# Get AMD phase
amd_detection = self._heuristic_amd_detection(df_slice)
# Generate probability (use TPSL classifier if available)
if self._tpsl_classifier:
pred = self._tpsl_classifier.predict(features, rr_config)
prob = pred.get("prob_tp_first", 0.5)
confidence = pred.get("confidence", 0.5)
else:
# Heuristic probability based on momentum
returns_5 = df_slice['close'].pct_change(5).iloc[-1]
prob = 0.5 + (returns_5 * 5) # Simple momentum-based
prob = max(0.35, min(0.65, prob))
confidence = 0.5
# Determine direction based on AMD phase
if amd_detection.phase == AMDPhase.ACCUMULATION:
direction = 'long'
elif amd_detection.phase == AMDPhase.DISTRIBUTION:
direction = 'short'
else:
# Use momentum
direction = 'long' if returns_5 > 0 else 'short'
# Set signal values
idx = df.index[i]
signals.loc[idx, 'prob_tp_first'] = prob
signals.loc[idx, 'direction'] = direction
signals.loc[idx, 'amd_phase'] = amd_detection.phase.value
signals.loc[idx, 'confidence'] = confidence
except Exception as e:
logger.debug(f"Signal generation failed at bar {i}: {e}")
continue
return signals
async def generate_signal(
self,
symbol: str,