signal_engine.py

"""
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
        }