""" Signal Engine Core Logic Training/inference/audit separation with proper forward-only predictions """ import pandas as pd import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple import uuid import json import hashlib from database import ( init_db, get_session, Symbol, Candle, PredictionBatch, Prediction, LedgerEntry, compute_batch_hash, compute_ledger_entry ) from ml_ensemble import FeatureEngineer, MLEnsemble from gateio_client import GateIOClient from config import settings class SignalEngine: """Main signal engine with training/inference/audit separation""" def __init__(self): self.engine = init_db(settings.database_url) self.feature_engineer = FeatureEngineer( volatility_window=settings.volatility_window, rsi_period=settings.rsi_period ) self.ensemble = MLEnsemble() self.gateio_client = GateIOClient() async def initialize_symbols(self, min_price: float = None, min_volume: float = None): """Initialize and filter symbols from Gate.io""" min_price = min_price or settings.min_price min_volume = min_volume or settings.min_volume_24h async with GateIOClient() as client: symbols = await client.filter_symbols(min_price, min_volume) session = get_session(settings.database_url) for symbol_data in symbols: existing = session.query(Symbol).filter_by(symbol=symbol_data['symbol']).first() if not existing: symbol = Symbol( symbol=symbol_data['symbol'], name=symbol_data.get('name', ''), min_price=min_price, min_volume_24h=min_volume, active=True ) session.add(symbol) session.commit() session.close() return len(symbols) async def backfill_candles(self, days: int = 30): """Backfill candle data for all active symbols""" session = get_session(settings.database_url) symbols = session.query(Symbol).filter_by(active=True).all() session.close() async with GateIOClient() as client: for symbol in symbols: df = await client.backfill_candles(symbol.symbol, days=days) if not df.empty: session = get_session(settings.database_url) for _, row in df.iterrows(): existing = session.query(Candle).filter_by( symbol=symbol.symbol, timestamp=row['timestamp'] ).first() if not existing: candle = Candle( symbol=symbol.symbol, timestamp=row['timestamp'], open=row['open'], high=row['high'], low=row['low'], close=row['close'], volume=row['volume'], quote_volume=row['quote_volume'] ) session.add(candle) session.commit() session.close() return True def build_training_dataset(self, symbol: str = None) -> pd.DataFrame: """Build training dataset with fwd_ret labels""" session = get_session(settings.database_url) query = session.query(Candle) if symbol: query = query.filter_by(symbol=symbol) candles = query.order_by(Candle.timestamp).all() session.close() if not candles: return pd.DataFrame() df = pd.DataFrame([{ 'symbol': c.symbol, 'timestamp': c.timestamp, 'open': c.open, 'high': c.high, 'low': c.low, 'close': c.close, 'volume': c.volume, 'quote_volume': c.quote_volume } for c in candles]) # Build features in training mode (requires fwd_ret) df = self.feature_engineer.build_features(df, mode='training') return df def build_inference_dataset(self, symbol: str = None) -> pd.DataFrame: """Build inference dataset without fwd_ret requirement""" session = get_session(settings.database_url) query = session.query(Candle) if symbol: query = query.filter_by(symbol=symbol) candles = query.order_by(Candle.timestamp).all() session.close() if not candles: return pd.DataFrame() df = pd.DataFrame([{ 'symbol': c.symbol, 'timestamp': c.timestamp, 'open': c.open, 'high': c.high, 'low': c.low, 'close': c.close, 'volume': c.volume, 'quote_volume': c.quote_volume } for c in candles]) # Build features in inference mode (no fwd_ret needed) df = self.feature_engineer.build_features(df, mode='inference') return df def train_model(self, symbol: str = None) -> Dict: """Train ML ensemble on historical data""" df = self.build_training_dataset(symbol) if df.empty or len(df) < 100: return {'error': 'Insufficient training data'} # Get feature columns feature_cols = self.feature_engineer.get_feature_columns() X = df[feature_cols].values y = df['target'].values # Fit ensemble metrics = self.ensemble.fit(X, y) self.ensemble.feature_columns = feature_cols return { 'training_samples': len(df), 'feature_columns': feature_cols, **metrics } def generate_predictions(self, target_hour: datetime = None) -> Dict: """ Generate predictions for the next hour CRITICAL: Uses inference mode (no fwd_ret required) for latest candle """ if target_hour is None: target_hour = datetime.utcnow().replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) # Build inference dataset (includes latest candle without fwd_ret) df = self.build_inference_dataset() if df.empty: return {'error': 'No data available'} # Get latest row for each symbol (this is the key fix - no fwd_ret needed) latest_rows = df.groupby('symbol').last().reset_index() # Get features feature_cols = self.feature_engineer.get_feature_columns() X = latest_rows[feature_cols].values # Predict if not self.ensemble.fitted: return {'error': 'Model not trained'} predictions = self.ensemble.predict(X) # Build prediction batch batch_id = str(uuid.uuid4()) predictions_list = [] for i, row in latest_rows.iterrows(): pred = { 'symbol': row['symbol'], 'target_hour': target_hour.isoformat(), 'entry_price': row['close'], 'direction': 'LONG' if predictions['direction'][i] > 0 else 'SHORT', 'probability_up': float(predictions['probability_up'][i]), 'confidence': float(predictions['confidence'][i]), 'suggested_position': float(predictions['position'][i]), 'feature_hash': self.ensemble.compute_feature_hash(X[i:i+1]) } predictions_list.append(pred) # Compute batch hash (full commitment) metadata = { 'batch_id': batch_id, 'target_hour': target_hour.isoformat(), 'committed_at': datetime.utcnow().isoformat(), 'model_version': '1.0', 'ensemble_weights': self.ensemble.weights } batch_hash = compute_batch_hash(predictions_list, metadata) # Get previous batch hash for chain session = get_session(settings.database_url) prev_batch = session.query(PredictionBatch).order_by(PredictionBatch.id.desc()).first() prev_batch_hash = prev_batch.batch_hash if prev_batch else None # Store in database batch = PredictionBatch( batch_id=batch_id, target_hour=target_hour, batch_hash=batch_hash, prev_batch_hash=prev_batch_hash, model_version='1.0', ensemble_weights=self.ensemble.weights, num_predictions=len(predictions_list), avg_confidence=sum(p['confidence'] for p in predictions_list) / len(predictions_list) ) session.add(batch) for pred in predictions_list: prediction = Prediction( batch_id=batch_id, symbol=pred['symbol'], target_hour=target_hour, entry_price=pred['entry_price'], direction=pred['direction'], probability_up=pred['probability_up'], confidence=pred['confidence'], suggested_position=pred['suggested_position'], feature_hash=pred['feature_hash'] ) session.add(prediction) # Ledger entry for batch commitment ledger_payload = { 'batch_id': batch_id, 'target_hour': target_hour.isoformat(), 'predictions': predictions_list, 'metadata': metadata } entry_hash, payload_hash, canonical = compute_ledger_entry( 'BATCH_COMMIT', ledger_payload, prev_batch_hash ) ledger = LedgerEntry( entry_hash=entry_hash, prev_hash=prev_batch_hash, entry_type='BATCH_COMMIT', payload=canonical, payload_hash=payload_hash ) session.add(ledger) session.commit() session.close() return { 'batch_id': batch_id, 'target_hour': target_hour.isoformat(), 'batch_hash': batch_hash, 'num_predictions': len(predictions_list), 'predictions': predictions_list[:50], # Preview first 50 'full_predictions': predictions_list # Full list for audit } def score_predictions(self, target_hour: datetime = None) -> Dict: """Score predictions after target hour closes""" if target_hour is None: target_hour = datetime.utcnow().replace(minute=0, second=0, microsecond=0) - timedelta(hours=1) session = get_session(settings.database_url) # Get unscored predictions for target hour predictions = session.query(Prediction).filter( Prediction.target_hour == target_hour, Prediction.scored_at.is_(None) ).all() if not predictions: session.close() return {'error': 'No predictions to score'} scored_count = 0 correct_count = 0 for pred in predictions: # Get exit price (next candle close) exit_candle = session.query(Candle).filter( Candle.symbol == pred.symbol, Candle.timestamp >= pred.target_hour + timedelta(hours=1) ).first() if exit_candle: pred.exit_price = exit_candle.close pred.actual_return = (exit_candle.close - pred.entry_price) / pred.entry_price pred.correct = (pred.actual_return > 0) == (pred.direction == 'LONG') pred.scored_at = datetime.utcnow() scored_count += 1 if pred.correct: correct_count += 1 # Update batch metrics batch = session.query(PredictionBatch).filter_by(target_hour=target_hour).first() if batch: batch.scored_at = datetime.utcnow() batch.accuracy = correct_count / scored_count if scored_count > 0 else 0 # Compute Brier score all_preds = session.query(Prediction).filter_by(batch_id=batch.batch_id).all() if all_preds: brier = sum((p.probability_up - (1 if p.correct else 0))**2 for p in all_preds) / len(all_preds) batch.brier_score = brier # Hypothetical PnL (simplified) hypothetical_pnl = sum( p.actual_return * abs(p.suggested_position) * 10000 for p in all_preds if p.actual_return is not None ) / len(all_preds) if all_preds else 0 batch.hypothetical_pnl_bps = hypothetical_pnl - settings.maker_fee_bps # Ledger entry for batch scoring prev_ledger = session.query(LedgerEntry).order_by(LedgerEntry.id.desc()).first() prev_hash = prev_ledger.entry_hash if prev_ledger else None ledger_payload = { 'target_hour': target_hour.isoformat(), 'scored_at': datetime.utcnow().isoformat(), 'scored_count': scored_count, 'correct_count': correct_count, 'accuracy': batch.accuracy if batch else 0 } entry_hash, payload_hash, canonical = compute_ledger_entry( 'BATCH_SCORE', ledger_payload, prev_hash ) ledger = LedgerEntry( entry_hash=entry_hash, prev_hash=prev_hash, entry_type='BATCH_SCORE', payload=canonical, payload_hash=payload_hash ) session.add(ledger) session.commit() session.close() return { 'target_hour': target_hour.isoformat(), 'scored_count': scored_count, 'correct_count': correct_count, 'accuracy': correct_count / scored_count if scored_count > 0 else 0 }