trading-platform/docs/99-analisis/ET-REFACTORING-MINIMO-VIABLE.md

title	version	date	status	author	epic	tags
Especificación Técnica - Refactoring Mínimo Viable	1.0.0	2026-01-06	Proposed	ML-Specialist + Orquestador	OQI-006	refactoring integration ml attention

REFACTORING MÍNIMO VIABLE - ML PREDICTION SYSTEM

1. RESUMEN EJECUTIVO

Este documento define el plan de refactoring de bajo riesgo para integrar la infraestructura ML existente y lograr mejoras inmediatas en win rate y R:R ratio.

Objetivo: Pasar de 33-44% win rate a 60%+ con cambios mínimos.

Principio: NO reescribir, INTEGRAR código existente.

---

2. CAMBIOS PROPUESTOS

2.1 CAMBIO 1: Cargar Modelos Entrenados (Prioridad ALTA)

Archivo: src/services/prediction_service.py

Antes (línea ~157):

from ..models.range_predictor import RangePredictor
self._range_predictor = RangePredictor()  # Modelo genérico

Después:

from ..training.symbol_timeframe_trainer import SymbolTimeframeTrainer

class PredictionService:
    def __init__(self):
        self._trainers: Dict[str, SymbolTimeframeTrainer] = {}
        self._load_trained_models()

    def _load_trained_models(self):
        """Cargar modelos entrenados por símbolo/timeframe"""
        models_path = Path(__file__).parent.parent.parent / 'models' / 'ml_first'

        for symbol_dir in models_path.iterdir():
            if symbol_dir.is_dir():
                symbol = symbol_dir.name
                trainer = SymbolTimeframeTrainer()
                trainer.load(str(symbol_dir))
                self._trainers[symbol] = trainer
                logger.info(f"✅ Loaded models for {symbol}")

    def predict_range(self, symbol: str, timeframe: str, df: pd.DataFrame):
        """Usar modelo específico por símbolo"""
        if symbol in self._trainers:
            return self._trainers[symbol].predict(df, symbol, timeframe)
        else:
            # Fallback a modelo legacy
            return self._range_predictor.predict(df)

Impacto Estimado: +5-10% precisión Riesgo: Bajo (fallback a legacy) Esfuerzo: 2 horas

---

2.2 CAMBIO 2: Eliminar Factores Hardcodeados (Prioridad ALTA)

Archivo: src/models/range_predictor_factor.py

Antes (línea 598-601):

class PriceDataGenerator:
    SYMBOLS = {
        'XAUUSD': {'base': 2650.0, 'volatility': 0.0012, 'factor': 2.5},
        'EURUSD': {'base': 1.0420, 'volatility': 0.0004, 'factor': 0.0003},
    }

Después:

from ..training.symbol_timeframe_trainer import SYMBOL_CONFIGS

class PriceDataGenerator:
    def __init__(self, symbol: str, seed: int = 42):
        self.symbol = symbol
        # Usar configuración centralizada
        config = SYMBOL_CONFIGS.get(symbol)
        if config:
            self.config = {
                'base': 2650.0 if symbol == 'XAUUSD' else 1.0,  # Precio actual dinámico
                'volatility': config.base_factor / 1000,  # Normalizar
                'factor': config.base_factor
            }
        else:
            # Default para símbolos nuevos
            self.config = self._compute_dynamic_config(symbol)

Impacto: Soporte para 5+ símbolos (vs 2) Riesgo: Bajo Esfuerzo: 1 hora

---

2.3 CAMBIO 3: Activar Filtros Direccionales (Prioridad ALTA)

Archivo: src/models/signal_generator.py

Agregar filtros basados en backtests exitosos:

class DirectionalFilters:
    """Filtros direccionales validados en backtests"""

    @staticmethod
    def is_short_valid(indicators: Dict, symbol: str) -> Tuple[bool, int]:
        """
        Validar señal SHORT (2+ confirmaciones)

        Returns:
            (is_valid, confirmation_count)
        """
        confirmations = 0

        # Filtros que funcionaron en XAUUSD
        if indicators.get('rsi', 50) > 55:
            confirmations += 1
        if indicators.get('sar_above_price', False):
            confirmations += 1
        if indicators.get('cmf', 0) < 0:
            confirmations += 1
        if indicators.get('mfi', 50) > 55:
            confirmations += 1

        return confirmations >= 2, confirmations

    @staticmethod
    def is_long_valid(indicators: Dict, symbol: str) -> Tuple[bool, int]:
        """
        Validar señal LONG (3+ confirmaciones, más estricto)
        """
        confirmations = 0

        if indicators.get('rsi', 50) < 35:
            confirmations += 1
        if not indicators.get('sar_above_price', True):
            confirmations += 1
        if indicators.get('cmf', 0) > 0.1:
            confirmations += 1
        if indicators.get('mfi', 50) < 35:
            confirmations += 1

        return confirmations >= 3, confirmations


# En SignalGenerator.generate()
def generate(self, df: pd.DataFrame, symbol: str, ...):
    # ... código existente ...

    # Aplicar filtros direccionales
    indicators = self._compute_indicators(df)

    if direction == Direction.SHORT:
        is_valid, conf_count = DirectionalFilters.is_short_valid(indicators, symbol)
        if not is_valid:
            return self._neutral_signal()
        confidence *= (1 + 0.1 * conf_count)  # Boost por confirmaciones

    elif direction == Direction.LONG:
        is_valid, conf_count = DirectionalFilters.is_long_valid(indicators, symbol)
        if not is_valid:
            return self._neutral_signal()
        confidence *= (1 + 0.1 * conf_count)

    # ... resto del código ...

Impacto: +10-15% win rate (demostrado en backtests) Riesgo: Bajo (solo filtra, no cambia lógica) Esfuerzo: 3 horas

---

2.4 CAMBIO 4: Integrar DynamicFactorWeighter (Prioridad MEDIA)

Archivo: src/models/enhanced_range_predictor.py y otros

Agregar attention weighting a XGBoost:

from ..training.dynamic_factor_weighting import DynamicFactorWeighter, DynamicFactorConfig

class AttentionWeightedPredictor:
    """Wrapper para agregar attention a cualquier modelo"""

    def __init__(self, base_model, config: DynamicFactorConfig = None):
        self.base_model = base_model
        self.weighter = DynamicFactorWeighter(config or DynamicFactorConfig())

    def fit(self, X: np.ndarray, y: np.ndarray, df: pd.DataFrame):
        """Entrenar con sample weights de atención"""
        # Calcular pesos de atención
        weights = self.weighter.compute_weights(df)

        # Entrenar modelo base con pesos
        self.base_model.fit(X, y, sample_weight=weights)

        return self

    def predict(self, X: np.ndarray, df: pd.DataFrame = None):
        """Predecir (opcional: devolver attention weight)"""
        predictions = self.base_model.predict(X)

        if df is not None:
            weights = self.weighter.compute_weights(df, normalize=False)
            return predictions, weights

        return predictions

Impacto: +5-10% enfoque en movimientos significativos Riesgo: Medio (requiere reentrenamiento) Esfuerzo: 4 horas

---

3. ORDEN DE IMPLEMENTACIÓN

Semana 1:
├── Día 1-2: CAMBIO 3 - Filtros direccionales
│   └── Test: Backtesting XAUUSD con filtros
├── Día 3-4: CAMBIO 1 - Cargar modelos entrenados
│   └── Test: Verificar predicciones por símbolo
└── Día 5: CAMBIO 2 - Eliminar hardcoding
    └── Test: Agregar BTCUSD como nuevo símbolo

Semana 2:
├── Día 1-3: CAMBIO 4 - Integrar DynamicFactorWeighter
│   └── Test: Reentrenar modelo XAUUSD 5m con attention
└── Día 4-5: Backtesting completo + ajustes

---

4. TESTS DE REGRESIÓN

4.1 Tests Unitarios

# tests/test_prediction_integration.py

def test_symbol_specific_model_loaded():
    """Verificar que se cargan modelos por símbolo"""
    service = PredictionService()
    assert 'XAUUSD' in service._trainers
    assert service._trainers['XAUUSD'] is not None

def test_directional_filters_short():
    """Verificar filtros SHORT"""
    indicators = {'rsi': 60, 'sar_above_price': True, 'cmf': -0.1, 'mfi': 60}
    is_valid, count = DirectionalFilters.is_short_valid(indicators, 'XAUUSD')
    assert is_valid == True
    assert count >= 2

def test_directional_filters_long():
    """Verificar filtros LONG (más estrictos)"""
    indicators = {'rsi': 30, 'sar_above_price': False, 'cmf': 0.15, 'mfi': 30}
    is_valid, count = DirectionalFilters.is_long_valid(indicators, 'XAUUSD')
    assert is_valid == True
    assert count >= 3

def test_attention_weights_computation():
    """Verificar cálculo de attention weights"""
    df = create_sample_ohlcv(n=500)
    weighter = DynamicFactorWeighter()
    weights = weighter.compute_weights(df)

    assert len(weights) == len(df)
    assert weights.mean() > 0
    assert weights.max() <= 3.0  # w_max

def test_fallback_to_legacy():
    """Verificar fallback para símbolos no entrenados"""
    service = PredictionService()
    df = create_sample_ohlcv()

    # Símbolo no entrenado
    result = service.predict_range('UNKNOWN', '5m', df)
    assert result is not None  # Fallback funciona

4.2 Tests de Integración

# tests/test_backtesting_regression.py

def test_xauusd_5m_win_rate():
    """Verificar win rate no disminuye"""
    results = run_backtest('XAUUSD', '5m', period='2025-01')

    # Baseline: 44% con filtros actuales
    assert results['win_rate'] >= 0.40

def test_xauusd_5m_profit_factor():
    """Verificar profit factor"""
    results = run_backtest('XAUUSD', '5m', period='2025-01')

    # Baseline: 1.07
    assert results['profit_factor'] >= 1.0

def test_attention_improves_signal_quality():
    """Verificar que attention mejora selección"""
    # Sin attention
    signals_no_attn = generate_signals(use_attention=False)

    # Con attention
    signals_with_attn = generate_signals(use_attention=True)

    # Debe haber menos señales pero mejor calidad
    assert len(signals_with_attn) <= len(signals_no_attn)
    assert signals_with_attn.mean_confidence >= signals_no_attn.mean_confidence

4.3 Tests de Performance

# tests/test_performance.py

def test_prediction_latency():
    """Verificar latencia de predicción < 100ms"""
    service = PredictionService()
    df = create_sample_ohlcv(n=500)

    start = time.time()
    for _ in range(100):
        service.predict_range('XAUUSD', '5m', df)
    elapsed = (time.time() - start) / 100

    assert elapsed < 0.1  # < 100ms por predicción

def test_model_loading_time():
    """Verificar tiempo de carga < 5 segundos"""
    start = time.time()
    service = PredictionService()
    elapsed = time.time() - start

    assert elapsed < 5.0

---

5. CRITERIOS DE ÉXITO

Métrica	Baseline	Post-Refactoring	Meta Final
Win Rate	33-44%	≥ 55%	80%
Profit Factor	1.07	≥ 1.2	1.8
R:R Ratio	1.2:1	≥ 1.8:1	2.5:1
Latencia	50ms	< 100ms	< 50ms
Símbolos	2-3	5+	100+

---

6. ROLLBACK PLAN

En caso de regresión:

1. Rollback inmediato: Revertir a rama main anterior
2. Fallback en código: Cada cambio tiene fallback a comportamiento legacy
3. Feature flags:

   USE_TRAINED_MODELS = os.getenv('USE_TRAINED_MODELS', 'true') == 'true'
   USE_DIRECTIONAL_FILTERS = os.getenv('USE_DIRECTIONAL_FILTERS', 'true') == 'true'
   USE_ATTENTION_WEIGHTING = os.getenv('USE_ATTENTION_WEIGHTING', 'false') == 'true'
   

---

7. MONITOREO POST-DEPLOY

7.1 Métricas a Monitorear

# Agregar a prediction_service.py
from prometheus_client import Counter, Histogram

PREDICTIONS_TOTAL = Counter('ml_predictions_total', 'Total predictions', ['symbol', 'direction'])
PREDICTION_LATENCY = Histogram('ml_prediction_latency_seconds', 'Prediction latency')
SIGNALS_FILTERED = Counter('ml_signals_filtered', 'Signals filtered by direction', ['reason'])

7.2 Alertas

Alerta	Condición	Acción
Win Rate Drop	win_rate < 0.40 por 24h	Review filtros
Latency High	p99 > 200ms	Check model loading
No Signals	0 signals en 8h	Check filters/data

---

Documento generado: 2026-01-06 Estado: Propuesto para revisión