156 lines
No EOL
4.7 KiB
Python
156 lines
No EOL
4.7 KiB
Python
# 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) |