Article-22220-Real-Time-Entropy-Trading-System/Server.py

# Copyright 2025, MetaQuotes Ltd.
# https://www.mql5.com/en/users/johnhlomohang/

from flask import Flask, request, jsonify
import numpy as np
import torch
import joblib
from collections import deque
import time
import json

from Model import EntropyModel
from Features import build_features, VolatilityRegimeDetector
import logging
log = logging.getLogger('werkzeug')
log.setLevel(logging.ERROR)  # Only show errors, not every request

app = Flask(__name__)

# Load model and scaler
try:
    model = EntropyModel()
    model.load_state_dict(torch.load("entropy_model.pth", map_location='cpu'))
    model.eval()
    print("Model loaded")
except:
    model = None
    print("Model not loaded")

try:
    scaler = joblib.load("scaler.pkl")
    print("Scaler loaded")
except:
    scaler = None
    print("Scaler not loaded")

# Fast regime detector with shorter history for tick-level responsiveness
regime_detector = VolatilityRegimeDetector(window_size=50, history_size=50)

# Track entropy for delta calculation
entropy_history = deque(maxlen=10)
last_entropy = None

@app.route('/predict', methods=['POST'])
def predict():
    global last_entropy
    
    try:
        data = request.json
        prices = np.array(data["prices"], dtype=np.float32)
        rsi = float(data.get("rsi", 50.0))
        
        # Fast feature calculation
        features, metrics = build_features(prices, rsi, None, None)
        
        # Track entropy for delta
        current_entropy = metrics['entropy']
        delta_entropy = 0.0 if last_entropy is None else current_entropy - last_entropy
        last_entropy = current_entropy
        entropy_history.append(current_entropy)
        
        # Detect regime
        regime_info = regime_detector.update(metrics)
        
        # Scale and predict
        if scaler is not None and model is not None:
            scaled = scaler.transform([features])
            x = torch.tensor(scaled, dtype=torch.float32)
            with torch.no_grad():
                prob = model(x).item()
        else:
            prob = 0.5
        
        # Adaptive signal with entropy momentum
        signal = generate_adaptive_signal(prob, metrics, regime_info, delta_entropy)
        confidence = calculate_confidence(prob, metrics, regime_info, delta_entropy)
        
        return jsonify({
            "probability": float(prob),
            "entropy": float(metrics['entropy']),
            "vol_entropy": float(metrics['vol_entropy']),
            "delta_entropy": float(delta_entropy),
            "signal": signal,
            "regime": regime_info['regime'],
            "volatility_multiplier": float(regime_info['volatility_multiplier']),
            "confidence": float(confidence),
            "entropy_momentum": float(calculate_entropy_momentum())
        })
        
    except Exception as e:
        return jsonify({"signal": "HOLD", "error": str(e)}), 200


def generate_adaptive_signal(prob, metrics, regime_info, delta_entropy):
    """Fast adaptive signal generation for tick-level trading."""
    entropy = metrics['entropy']
    regime = regime_info['regime']
    
    # Base thresholds
    buy_threshold = 0.60
    sell_threshold = 0.40
    
    # Adjust for regime
    if regime == "HIGH_VOLATILITY":
        buy_threshold = 0.65
        sell_threshold = 0.35
    elif regime == "LOW_VOLATILITY":
        buy_threshold = 0.55
        sell_threshold = 0.45
    
    # Entropy momentum adjustment
    if delta_entropy > 0.02:  # Increasing entropy = increasing uncertainty
        buy_threshold += 0.05
        sell_threshold -= 0.05
    
    if prob > buy_threshold:
        return "BUY"
    elif prob < sell_threshold:
        return "SELL"
    else:
        return "HOLD"


def calculate_confidence(prob, metrics, regime_info, delta_entropy):
    """Confidence score with entropy penalty."""
    base_conf = abs(prob - 0.5) * 2
    
    # Penalize high entropy
    entropy_penalty = metrics['entropy'] * 0.3
    
    # Penalize increasing entropy
    if delta_entropy > 0:
        entropy_penalty += delta_entropy * 2
    
    confidence = base_conf - entropy_penalty
    return max(0.0, min(1.0, confidence))


def calculate_entropy_momentum():
    """Calculate entropy momentum from history."""
    if len(entropy_history) < 3:
        return 0.0
    hist = list(entropy_history)
    return hist[-1] - hist[0]


@app.route('/health', methods=['GET'])
def health():
    return jsonify({"status": "ready"})


if __name__ == "__main__":
    print("\n" + "="*50)
    print("TICK-LEVEL ENTROPY SERVER")
    print("="*50)
    app.run(host="127.0.0.1", port=5000, debug=False, threaded=True)