Phạm Phú Nguyễn Hưng
4cb86ecd5f
- 9 indicator signal modes: Ichimoku, EMA Cross, RSI, BB Bounce, Stoch, CCI, MACD, Supertrend, Momentum - DCA multi-tier with adaptive distance and lot multiplier - CCBSN (partial close 50% -> breakeven -> trailing) - Sniper (trim oldest losing orders) - Anti-Detect for Prop Firm compliance - Python ML optimizer for DCA parameters - Wave strategy brute-force optimizer (2240+ combos) - 85+ optimizable inputs for MT5 Strategy Tester - Auto-detect filling type (Exness compatibility) - Retry logic for order closing Built by @hungpixi | Comarai.com
878 lines
36 KiB
Python
878 lines
36 KiB
Python
"""
|
||
DCA ML Optimizer - Machine Learning DCA Parameter Optimization
|
||
================================================================
|
||
Phân tích dữ liệu nến lịch sử XAUUSD, tìm khoảng cách DCA + lot tối ưu
|
||
theo volatility regime, phát hiện pattern "blow-up".
|
||
|
||
Usage:
|
||
python dca_ml_optimizer.py --data XAUUSD_M5.csv
|
||
python dca_ml_optimizer.py --data XAUUSD_M5.csv --export-mt5
|
||
python dca_ml_optimizer.py --test (chạy với sample data)
|
||
|
||
Output:
|
||
- ml_params.mqh (hardcoded params cho MQ5)
|
||
- optimization_log.csv (log chi tiết)
|
||
|
||
Author: Comarai (https://comarai.com) - AI-powered trading optimization
|
||
"""
|
||
|
||
import argparse
|
||
import os
|
||
import sys
|
||
import csv
|
||
import json
|
||
import math
|
||
from datetime import datetime, timedelta
|
||
from collections import defaultdict
|
||
|
||
import numpy as np
|
||
|
||
# Try import sklearn, fallback to simple heuristic if not available
|
||
try:
|
||
from sklearn.tree import DecisionTreeRegressor
|
||
from sklearn.ensemble import GradientBoostingRegressor
|
||
from sklearn.model_selection import cross_val_score
|
||
HAS_SKLEARN = True
|
||
except ImportError:
|
||
HAS_SKLEARN = False
|
||
print("[WARN] scikit-learn not installed. Using heuristic optimization.")
|
||
print(" Install: pip install scikit-learn numpy pandas")
|
||
|
||
|
||
# =============================================================================
|
||
# CONSTANTS
|
||
# =============================================================================
|
||
PIP_VALUE_XAUUSD = 0.1 # 1 pip = 0.1 cho XAUUSD (2 or 3 digit broker)
|
||
|
||
# Default settings from 2-2-test.set (reference)
|
||
DEFAULT_DCA_DISTANCE = 10.0 # pips
|
||
DEFAULT_DCA_DIST_MULTI = 1.2
|
||
DEFAULT_LOT = 0.19
|
||
DEFAULT_LOT_MULTI = 1.0 # lot cố định
|
||
DEFAULT_MAX_DCA = 5
|
||
DEFAULT_TP_DCA = 50.0 # pips
|
||
|
||
# Regime thresholds (will be refined by ML)
|
||
ATR_PERCENTILES = [25, 50, 75, 90] # Low, Medium, High, Extreme
|
||
|
||
|
||
# =============================================================================
|
||
# DATA LOADING
|
||
# =============================================================================
|
||
def load_candle_data(filepath):
|
||
"""Load candle data from CSV exported by MT5.
|
||
|
||
Supports formats:
|
||
- MT5 default export: Date, Time, Open, High, Low, Close, TickVolume, Volume, Spread
|
||
- Custom: datetime, open, high, low, close, volume
|
||
"""
|
||
candles = []
|
||
|
||
with open(filepath, 'r', encoding='utf-8-sig') as f:
|
||
# Detect delimiter
|
||
first_line = f.readline()
|
||
f.seek(0)
|
||
|
||
delimiter = '\t' if '\t' in first_line else ','
|
||
reader = csv.reader(f, delimiter=delimiter)
|
||
|
||
# Try to detect header
|
||
header = next(reader)
|
||
header_lower = [h.strip().lower() for h in header]
|
||
|
||
# Map columns
|
||
col_map = {}
|
||
for i, h in enumerate(header_lower):
|
||
if h in ('date', 'datetime', '<date>'):
|
||
col_map['date'] = i
|
||
elif h in ('time', '<time>'):
|
||
col_map['time'] = i
|
||
elif h in ('open', '<open>'):
|
||
col_map['open'] = i
|
||
elif h in ('high', '<high>'):
|
||
col_map['high'] = i
|
||
elif h in ('low', '<low>'):
|
||
col_map['low'] = i
|
||
elif h in ('close', '<close>'):
|
||
col_map['close'] = i
|
||
elif h in ('tickvol', 'tick_volume', '<tickvol>', 'volume', '<vol>'):
|
||
col_map['volume'] = i
|
||
elif h in ('spread', '<spread>'):
|
||
col_map['spread'] = i
|
||
|
||
if 'open' not in col_map:
|
||
raise ValueError(f"Cannot detect OHLC columns. Header: {header}")
|
||
|
||
for row in reader:
|
||
try:
|
||
if len(row) < 4:
|
||
continue
|
||
|
||
candle = {
|
||
'open': float(row[col_map['open']]),
|
||
'high': float(row[col_map['high']]),
|
||
'low': float(row[col_map['low']]),
|
||
'close': float(row[col_map['close']]),
|
||
}
|
||
|
||
# Parse datetime
|
||
if 'date' in col_map and 'time' in col_map:
|
||
date_str = row[col_map['date']].strip()
|
||
time_str = row[col_map['time']].strip()
|
||
for fmt in ('%Y.%m.%d %H:%M:%S', '%Y.%m.%d %H:%M',
|
||
'%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M',
|
||
'%Y/%m/%d %H:%M:%S', '%Y/%m/%d %H:%M'):
|
||
try:
|
||
candle['datetime'] = datetime.strptime(f"{date_str} {time_str}", fmt)
|
||
break
|
||
except ValueError:
|
||
continue
|
||
elif 'date' in col_map:
|
||
date_str = row[col_map['date']].strip()
|
||
for fmt in ('%Y.%m.%d %H:%M:%S', '%Y.%m.%d %H:%M',
|
||
'%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M',
|
||
'%Y/%m/%d %H:%M:%S', '%Y/%m/%d %H:%M',
|
||
'%Y.%m.%d', '%Y-%m-%d'):
|
||
try:
|
||
candle['datetime'] = datetime.strptime(date_str, fmt)
|
||
break
|
||
except ValueError:
|
||
continue
|
||
|
||
if 'volume' in col_map:
|
||
candle['volume'] = int(float(row[col_map['volume']]))
|
||
if 'spread' in col_map:
|
||
candle['spread'] = int(float(row[col_map['spread']]))
|
||
|
||
candles.append(candle)
|
||
except (ValueError, IndexError):
|
||
continue
|
||
|
||
print(f"[INFO] Loaded {len(candles)} candles from {filepath}")
|
||
if candles:
|
||
print(f" Range: {candles[0].get('datetime', 'N/A')} -> {candles[-1].get('datetime', 'N/A')}")
|
||
return candles
|
||
|
||
|
||
def generate_sample_data(n=50000):
|
||
"""Generate realistic XAUUSD M5 sample data for testing."""
|
||
np.random.seed(42)
|
||
candles = []
|
||
price = 2650.0 # Starting price
|
||
dt = datetime(2024, 1, 1, 0, 0)
|
||
|
||
# Simulate different volatility regimes
|
||
regime_length = 2000 # ~7 days of M5 candles
|
||
|
||
for i in range(n):
|
||
# Change regime periodically
|
||
regime = (i // regime_length) % 4
|
||
if regime == 0: # Low volatility
|
||
vol = 0.8
|
||
elif regime == 1: # Medium volatility
|
||
vol = 2.0
|
||
elif regime == 2: # High volatility
|
||
vol = 4.0
|
||
else: # Extreme (news/blow-up)
|
||
vol = 8.0
|
||
|
||
# Random walk with drift
|
||
drift = np.random.normal(0, 0.01)
|
||
change = np.random.normal(drift, vol)
|
||
|
||
o = price
|
||
h = o + abs(np.random.normal(0, vol * 0.7))
|
||
l = o - abs(np.random.normal(0, vol * 0.7))
|
||
c = o + change
|
||
|
||
h = max(h, o, c) + abs(np.random.normal(0, vol * 0.2))
|
||
l = min(l, o, c) - abs(np.random.normal(0, vol * 0.2))
|
||
|
||
candles.append({
|
||
'datetime': dt,
|
||
'open': round(o, 2),
|
||
'high': round(h, 2),
|
||
'low': round(l, 2),
|
||
'close': round(c, 2),
|
||
'volume': np.random.randint(100, 5000),
|
||
'spread': np.random.randint(20, 50),
|
||
})
|
||
|
||
price = c
|
||
dt += timedelta(minutes=5)
|
||
# Skip weekends
|
||
if dt.weekday() >= 5:
|
||
dt += timedelta(days=7 - dt.weekday())
|
||
|
||
print(f"[INFO] Generated {len(candles)} sample candles")
|
||
return candles
|
||
|
||
|
||
# =============================================================================
|
||
# FEATURE ENGINEERING
|
||
# =============================================================================
|
||
def compute_atr(candles, period=14):
|
||
"""Compute ATR (Average True Range) in price terms."""
|
||
atrs = [0.0] * len(candles)
|
||
|
||
for i in range(1, len(candles)):
|
||
tr = max(
|
||
candles[i]['high'] - candles[i]['low'],
|
||
abs(candles[i]['high'] - candles[i-1]['close']),
|
||
abs(candles[i]['low'] - candles[i-1]['close'])
|
||
)
|
||
|
||
if i < period:
|
||
atrs[i] = tr
|
||
else:
|
||
atrs[i] = (atrs[i-1] * (period - 1) + tr) / period
|
||
|
||
return atrs
|
||
|
||
|
||
def compute_volatility_features(candles, atr_m5):
|
||
"""Compute volatility regime features for each candle."""
|
||
features = []
|
||
|
||
# ATR values for H1 and H4 aggregation
|
||
h1_candles_per = 12 # 12 M5 candles = 1 hour
|
||
h4_candles_per = 48 # 48 M5 candles = 4 hours
|
||
d1_candles_per = 288 # 288 M5 candles = 1 day
|
||
|
||
# Compute rolling ATR statistics
|
||
atr_lookback = 100 # 100 bars lookback for percentile
|
||
|
||
for i in range(max(d1_candles_per, atr_lookback), len(candles)):
|
||
atr_val = atr_m5[i]
|
||
|
||
# ATR percentile (where does current ATR sit vs recent history)
|
||
recent_atrs = atr_m5[i - atr_lookback:i]
|
||
recent_atrs_sorted = sorted(recent_atrs)
|
||
atr_percentile = sum(1 for x in recent_atrs_sorted if x <= atr_val) / len(recent_atrs_sorted) * 100
|
||
|
||
# H1 range (avg of last 12 candles)
|
||
h1_range = sum(c['high'] - c['low'] for c in candles[i - h1_candles_per:i]) / h1_candles_per
|
||
|
||
# H4 range
|
||
h4_range = sum(c['high'] - c['low'] for c in candles[i - h4_candles_per:i]) / h4_candles_per
|
||
|
||
# D1 range
|
||
d1_high = max(c['high'] for c in candles[i - d1_candles_per:i])
|
||
d1_low = min(c['low'] for c in candles[i - d1_candles_per:i])
|
||
d1_range = d1_high - d1_low
|
||
|
||
# Trend strength (simple: price vs 50-bar SMA)
|
||
sma50 = sum(c['close'] for c in candles[i - 50:i]) / 50
|
||
trend_strength = (candles[i]['close'] - sma50) / sma50 * 100 # % away from SMA
|
||
|
||
# Candle body ratio (body / total range)
|
||
body = abs(candles[i]['close'] - candles[i]['open'])
|
||
total_range = candles[i]['high'] - candles[i]['low']
|
||
body_ratio = body / total_range if total_range > 0 else 0
|
||
|
||
# Upper/lower wick ratio
|
||
if candles[i]['close'] >= candles[i]['open']:
|
||
upper_wick = candles[i]['high'] - candles[i]['close']
|
||
lower_wick = candles[i]['open'] - candles[i]['low']
|
||
else:
|
||
upper_wick = candles[i]['high'] - candles[i]['open']
|
||
lower_wick = candles[i]['close'] - candles[i]['low']
|
||
|
||
wick_ratio = upper_wick / lower_wick if lower_wick > 0 else 10.0
|
||
|
||
# Max drawdown in recent N candles (simulating worst case reversal)
|
||
max_up_move = 0
|
||
max_down_move = 0
|
||
for lookback in [24, 48, 96, 288]: # 2h, 4h, 8h, 24h
|
||
if i >= lookback:
|
||
start_price = candles[i - lookback]['close']
|
||
max_price = max(c['high'] for c in candles[i - lookback:i])
|
||
min_price = min(c['low'] for c in candles[i - lookback:i])
|
||
max_up_move = max(max_up_move, (max_price - start_price) / PIP_VALUE_XAUUSD)
|
||
max_down_move = max(max_down_move, (start_price - min_price) / PIP_VALUE_XAUUSD)
|
||
|
||
# Volatility regime classification
|
||
if atr_percentile < 25:
|
||
regime = 0 # LOW
|
||
elif atr_percentile < 50:
|
||
regime = 1 # MEDIUM
|
||
elif atr_percentile < 75:
|
||
regime = 2 # HIGH
|
||
else:
|
||
regime = 3 # EXTREME
|
||
|
||
features.append({
|
||
'idx': i,
|
||
'atr_m5': atr_val,
|
||
'atr_percentile': atr_percentile,
|
||
'h1_range': h1_range,
|
||
'h4_range': h4_range,
|
||
'd1_range': d1_range,
|
||
'trend_strength': trend_strength,
|
||
'body_ratio': body_ratio,
|
||
'wick_ratio': min(wick_ratio, 10.0),
|
||
'max_up_pips': max_up_move,
|
||
'max_down_pips': max_down_move,
|
||
'regime': regime,
|
||
'price': candles[i]['close'],
|
||
})
|
||
|
||
return features
|
||
|
||
|
||
# =============================================================================
|
||
# DCA SIMULATION
|
||
# =============================================================================
|
||
def simulate_dca_chain(candles, start_idx, direction, dca_distance, dca_dist_multi,
|
||
lot_size, lot_multi, max_dca, tp_pips, max_total_lots=50.0,
|
||
max_bars=5760): # max 5760 bars = 20 days
|
||
"""Simulate a DCA chain starting from start_idx.
|
||
|
||
Returns:
|
||
dict with result: 'tp_hit', 'blowup', 'timeout', and stats
|
||
"""
|
||
entry_price = candles[start_idx]['close']
|
||
|
||
# Track orders
|
||
orders = [{'price': entry_price, 'lots': lot_size, 'bar': start_idx}]
|
||
total_lots = lot_size
|
||
|
||
next_dca_dist = dca_distance
|
||
bars_elapsed = 0
|
||
max_adverse_pips = 0
|
||
max_lots_reached = total_lots
|
||
|
||
for i in range(start_idx + 1, min(start_idx + max_bars, len(candles))):
|
||
bars_elapsed += 1
|
||
current_price = candles[i]['close']
|
||
current_high = candles[i]['high']
|
||
current_low = candles[i]['low']
|
||
|
||
# Calculate average entry price
|
||
avg_price = sum(o['price'] * o['lots'] for o in orders) / total_lots
|
||
|
||
# Check TP
|
||
if direction == 'buy':
|
||
profit_pips = (current_high - avg_price) / PIP_VALUE_XAUUSD
|
||
adverse_pips = (avg_price - current_low) / PIP_VALUE_XAUUSD
|
||
else:
|
||
profit_pips = (avg_price - current_low) / PIP_VALUE_XAUUSD
|
||
adverse_pips = (current_high - avg_price) / PIP_VALUE_XAUUSD
|
||
|
||
max_adverse_pips = max(max_adverse_pips, adverse_pips)
|
||
|
||
if profit_pips >= tp_pips:
|
||
return {
|
||
'result': 'tp_hit',
|
||
'bars': bars_elapsed,
|
||
'total_lots': total_lots,
|
||
'max_lots': max_lots_reached,
|
||
'num_dca': len(orders) - 1,
|
||
'max_adverse_pips': max_adverse_pips,
|
||
'avg_price': avg_price,
|
||
'entry_price': entry_price,
|
||
}
|
||
|
||
# Check if should add DCA
|
||
last_price = orders[-1]['price']
|
||
if direction == 'buy':
|
||
dist_from_last = (last_price - current_price) / PIP_VALUE_XAUUSD
|
||
else:
|
||
dist_from_last = (current_price - last_price) / PIP_VALUE_XAUUSD
|
||
|
||
if dist_from_last >= next_dca_dist and len(orders) <= max_dca:
|
||
new_lot = lot_size
|
||
for _ in range(len(orders)):
|
||
new_lot *= lot_multi
|
||
new_lot = round(new_lot, 2)
|
||
|
||
if total_lots + new_lot > max_total_lots:
|
||
# BLOWUP: would exceed max lots
|
||
return {
|
||
'result': 'blowup',
|
||
'bars': bars_elapsed,
|
||
'total_lots': total_lots,
|
||
'max_lots': max_lots_reached,
|
||
'num_dca': len(orders) - 1,
|
||
'max_adverse_pips': max_adverse_pips,
|
||
'avg_price': avg_price,
|
||
'entry_price': entry_price,
|
||
'blow_distance_pips': adverse_pips,
|
||
}
|
||
|
||
orders.append({'price': current_price, 'lots': new_lot, 'bar': i})
|
||
total_lots += new_lot
|
||
max_lots_reached = max(max_lots_reached, total_lots)
|
||
next_dca_dist = dca_distance
|
||
for _ in range(len(orders) - 1):
|
||
next_dca_dist *= dca_dist_multi
|
||
|
||
# Timeout
|
||
avg_price = sum(o['price'] * o['lots'] for o in orders) / total_lots
|
||
return {
|
||
'result': 'timeout',
|
||
'bars': bars_elapsed,
|
||
'total_lots': total_lots,
|
||
'max_lots': max_lots_reached,
|
||
'num_dca': len(orders) - 1,
|
||
'max_adverse_pips': max_adverse_pips,
|
||
'avg_price': avg_price,
|
||
'entry_price': entry_price,
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# ML OPTIMIZATION
|
||
# =============================================================================
|
||
def optimize_dca_params(candles, features, lot_size=0.19):
|
||
"""Find optimal DCA parameters for each volatility regime."""
|
||
|
||
print("\n" + "=" * 60)
|
||
print(" DCA PARAMETER OPTIMIZATION")
|
||
print("=" * 60)
|
||
|
||
# Test different DCA distance settings per regime
|
||
distance_options = [5, 8, 10, 12, 15, 20, 25, 30, 40, 50]
|
||
max_dca_options = [2, 3, 4, 5, 6, 8]
|
||
lot_multi_options = [1.0, 1.1, 1.2, 1.3, 1.5]
|
||
|
||
# Sample starting points from features
|
||
sample_size = min(2000, len(features))
|
||
sample_indices = np.random.choice(len(features), sample_size, replace=False)
|
||
|
||
regime_results = {0: [], 1: [], 2: [], 3: []}
|
||
regime_names = {0: 'LOW_VOL', 1: 'MED_VOL', 2: 'HIGH_VOL', 3: 'EXTREME'}
|
||
|
||
print(f"\n[INFO] Testing {len(distance_options)} distances × {len(max_dca_options)} max_dca × {len(lot_multi_options)} lot_multi")
|
||
print(f" On {sample_size} sample starting points")
|
||
|
||
best_params = {}
|
||
|
||
for regime in range(4):
|
||
regime_samples = [features[i] for i in sample_indices if features[i]['regime'] == regime]
|
||
if len(regime_samples) < 20:
|
||
print(f"\n[WARN] Regime {regime_names[regime]}: only {len(regime_samples)} samples, skipping ML")
|
||
continue
|
||
|
||
print(f"\n--- Regime: {regime_names[regime]} ({len(regime_samples)} samples) ---")
|
||
|
||
best_score = -999
|
||
best_config = None
|
||
|
||
for dist in distance_options:
|
||
for max_d in max_dca_options:
|
||
for lot_m in lot_multi_options:
|
||
tp_hits = 0
|
||
blowups = 0
|
||
timeouts = 0
|
||
total_adverse = []
|
||
total_lots_used = []
|
||
|
||
# Test on subset of regime samples
|
||
test_samples = regime_samples[:min(100, len(regime_samples))]
|
||
|
||
for feat in test_samples:
|
||
idx = feat['idx']
|
||
if idx + 5760 >= len(candles):
|
||
continue
|
||
|
||
for direction in ['buy', 'sell']:
|
||
result = simulate_dca_chain(
|
||
candles, idx, direction,
|
||
dca_distance=dist,
|
||
dca_dist_multi=DEFAULT_DCA_DIST_MULTI,
|
||
lot_size=lot_size,
|
||
lot_multi=lot_m,
|
||
max_dca=max_d,
|
||
tp_pips=DEFAULT_TP_DCA,
|
||
max_total_lots=10.0,
|
||
)
|
||
|
||
if result['result'] == 'tp_hit':
|
||
tp_hits += 1
|
||
elif result['result'] == 'blowup':
|
||
blowups += 1
|
||
else:
|
||
timeouts += 1
|
||
|
||
total_adverse.append(result['max_adverse_pips'])
|
||
total_lots_used.append(result['max_lots'])
|
||
|
||
total = tp_hits + blowups + timeouts
|
||
if total == 0:
|
||
continue
|
||
|
||
# Score: maximize TP rate, minimize blowup rate, penalize high lots
|
||
tp_rate = tp_hits / total
|
||
blowup_rate = blowups / total
|
||
avg_adverse = np.mean(total_adverse) if total_adverse else 0
|
||
avg_lots = np.mean(total_lots_used) if total_lots_used else 0
|
||
|
||
# Weighted score: TP rate * 100 - blowup penalty - lots penalty
|
||
score = tp_rate * 100 - blowup_rate * 200 - avg_lots * 5 - avg_adverse * 0.1
|
||
|
||
if score > best_score:
|
||
best_score = score
|
||
best_config = {
|
||
'distance': dist,
|
||
'max_dca': max_d,
|
||
'lot_multi': lot_m,
|
||
'tp_rate': tp_rate,
|
||
'blowup_rate': blowup_rate,
|
||
'avg_adverse': avg_adverse,
|
||
'avg_lots': avg_lots,
|
||
'score': score,
|
||
'total_tests': total,
|
||
}
|
||
|
||
if best_config:
|
||
best_params[regime] = best_config
|
||
print(f" Best: dist={best_config['distance']}p, max_dca={best_config['max_dca']}, "
|
||
f"lot_multi={best_config['lot_multi']}")
|
||
print(f" Score: {best_config['score']:.1f} | TP: {best_config['tp_rate']*100:.1f}% | "
|
||
f"Blowup: {best_config['blowup_rate']*100:.1f}% | Avg adverse: {best_config['avg_adverse']:.1f}p")
|
||
|
||
regime_results[regime] = best_config
|
||
|
||
return best_params, regime_results
|
||
|
||
|
||
def detect_danger_patterns(candles, features):
|
||
"""Detect patterns that lead to blow-ups (31 lot sell scenario)."""
|
||
|
||
print("\n" + "=" * 60)
|
||
print(" DANGER PATTERN DETECTION")
|
||
print("=" * 60)
|
||
|
||
# Simulate with aggressive DCA settings to find blow-up scenarios
|
||
blowup_features = []
|
||
safe_features = []
|
||
|
||
sample_size = min(1000, len(features))
|
||
sample_indices = np.random.choice(len(features), sample_size, replace=False)
|
||
|
||
for idx_in_features in sample_indices:
|
||
feat = features[idx_in_features]
|
||
candle_idx = feat['idx']
|
||
if candle_idx + 5760 >= len(candles):
|
||
continue
|
||
|
||
for direction in ['buy', 'sell']:
|
||
result = simulate_dca_chain(
|
||
candles, candle_idx, direction,
|
||
dca_distance=10.0,
|
||
dca_dist_multi=1.2,
|
||
lot_size=0.19,
|
||
lot_multi=1.0,
|
||
max_dca=20, # Aggressive to find blowups
|
||
tp_pips=50.0,
|
||
max_total_lots=31.0, # Match the 31 lot scenario
|
||
)
|
||
|
||
feature_vec = [
|
||
feat['atr_m5'],
|
||
feat['atr_percentile'],
|
||
feat['h1_range'],
|
||
feat['h4_range'],
|
||
feat['d1_range'],
|
||
feat['trend_strength'],
|
||
feat['body_ratio'],
|
||
feat['wick_ratio'],
|
||
feat['max_up_pips'],
|
||
feat['max_down_pips'],
|
||
]
|
||
|
||
if result['result'] == 'blowup' or (result['result'] == 'timeout' and result['max_adverse_pips'] > 80):
|
||
blowup_features.append(feature_vec)
|
||
elif result['result'] == 'tp_hit':
|
||
safe_features.append(feature_vec)
|
||
|
||
print(f"\n Blowup scenarios found: {len(blowup_features)}")
|
||
print(f" Safe scenarios found: {len(safe_features)}")
|
||
|
||
# Find danger thresholds
|
||
danger_thresholds = {}
|
||
|
||
if blowup_features:
|
||
blow_arr = np.array(blowup_features)
|
||
safe_arr = np.array(safe_features) if safe_features else np.zeros((1, len(blowup_features[0])))
|
||
|
||
feature_names = ['atr_m5', 'atr_pct', 'h1_range', 'h4_range', 'd1_range',
|
||
'trend_str', 'body_ratio', 'wick_ratio', 'max_up', 'max_down']
|
||
|
||
print("\n Key danger indicators:")
|
||
for j, name in enumerate(feature_names):
|
||
blow_mean = np.mean(blow_arr[:, j])
|
||
blow_p75 = np.percentile(blow_arr[:, j], 75)
|
||
safe_mean = np.mean(safe_arr[:, j]) if len(safe_arr) > 1 else 0
|
||
|
||
if abs(blow_mean - safe_mean) > 0.01:
|
||
ratio = blow_mean / safe_mean if safe_mean != 0 else 999
|
||
print(f" {name}: blow={blow_mean:.2f} vs safe={safe_mean:.2f} (ratio: {ratio:.1f}x)")
|
||
danger_thresholds[name] = {
|
||
'blow_mean': blow_mean,
|
||
'blow_p75': blow_p75,
|
||
'safe_mean': safe_mean,
|
||
}
|
||
|
||
# Use ML to find danger threshold if sklearn available
|
||
if HAS_SKLEARN and len(safe_features) > 10:
|
||
X = np.vstack([blow_arr, safe_arr])
|
||
y = np.array([1] * len(blow_arr) + [0] * len(safe_arr))
|
||
|
||
clf = DecisionTreeRegressor(max_depth=3)
|
||
clf.fit(X, y)
|
||
|
||
importances = clf.feature_importances_
|
||
print("\n ML Feature Importance (danger prediction):")
|
||
for j, name in enumerate(feature_names):
|
||
if importances[j] > 0.05:
|
||
print(f" {name}: {importances[j]*100:.1f}%")
|
||
|
||
danger_thresholds['ml_model'] = clf
|
||
|
||
return danger_thresholds
|
||
|
||
|
||
# =============================================================================
|
||
# ATR REGIME THRESHOLDS
|
||
# =============================================================================
|
||
def compute_atr_thresholds(candles, features):
|
||
"""Compute ATR threshold values for regime classification in MQ5."""
|
||
all_atrs = [f['atr_m5'] for f in features]
|
||
|
||
thresholds = {
|
||
'atr_low': np.percentile(all_atrs, 25),
|
||
'atr_med': np.percentile(all_atrs, 50),
|
||
'atr_high': np.percentile(all_atrs, 75),
|
||
'atr_extreme': np.percentile(all_atrs, 90),
|
||
'atr_mean': np.mean(all_atrs),
|
||
}
|
||
|
||
print(f"\n ATR Thresholds (in price, XAUUSD):")
|
||
print(f" Low: < {thresholds['atr_low']:.2f}")
|
||
print(f" Medium: {thresholds['atr_low']:.2f} - {thresholds['atr_med']:.2f}")
|
||
print(f" High: {thresholds['atr_med']:.2f} - {thresholds['atr_high']:.2f}")
|
||
print(f" Extreme: > {thresholds['atr_high']:.2f}")
|
||
|
||
# Convert to pips for MQ5
|
||
pip_items = {k + '_pips': v / PIP_VALUE_XAUUSD for k, v in thresholds.items()}
|
||
thresholds.update(pip_items)
|
||
|
||
return thresholds
|
||
|
||
|
||
# =============================================================================
|
||
# OUTPUT GENERATION
|
||
# =============================================================================
|
||
def generate_mqh_file(best_params, atr_thresholds, danger_thresholds, output_path):
|
||
"""Generate ml_params.mqh with hardcoded optimal parameters."""
|
||
|
||
regime_names = {0: 'LOW_VOL', 1: 'MED_VOL', 2: 'HIGH_VOL', 3: 'EXTREME'}
|
||
|
||
# Defaults if regime not found
|
||
defaults = {
|
||
0: {'distance': 8, 'max_dca': 8, 'lot_multi': 1.3},
|
||
1: {'distance': 12, 'max_dca': 5, 'lot_multi': 1.2},
|
||
2: {'distance': 20, 'max_dca': 3, 'lot_multi': 1.0},
|
||
3: {'distance': 35, 'max_dca': 2, 'lot_multi': 1.0},
|
||
}
|
||
|
||
lines = []
|
||
lines.append("//+------------------------------------------------------------------+")
|
||
lines.append("//| ml_params.mqh - ML-Generated DCA Parameters |")
|
||
lines.append(f"//| Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} |")
|
||
lines.append("//| Author: Comarai (https://comarai.com) |")
|
||
lines.append("//+------------------------------------------------------------------+")
|
||
lines.append("#ifndef ML_PARAMS_MQH")
|
||
lines.append("#define ML_PARAMS_MQH")
|
||
lines.append("")
|
||
lines.append("// ===================================================================")
|
||
lines.append("// ATR REGIME THRESHOLDS (in PRICE, not pips)")
|
||
lines.append("// Computed from historical ATR percentiles")
|
||
lines.append("// ===================================================================")
|
||
lines.append(f"#define ML_ATR_LOW_THRESHOLD {atr_thresholds['atr_low']:.4f} // P25 ATR")
|
||
lines.append(f"#define ML_ATR_MED_THRESHOLD {atr_thresholds['atr_med']:.4f} // P50 ATR")
|
||
lines.append(f"#define ML_ATR_HIGH_THRESHOLD {atr_thresholds['atr_high']:.4f} // P75 ATR")
|
||
lines.append(f"#define ML_ATR_EXTREME_THRESHOLD {atr_thresholds['atr_extreme']:.4f} // P90 ATR")
|
||
lines.append("")
|
||
lines.append("// ===================================================================")
|
||
lines.append("// DCA DISTANCE PER REGIME (in pips)")
|
||
lines.append("// Wider distance when volatile = less DCA = less risk")
|
||
lines.append("// ===================================================================")
|
||
|
||
for regime in range(4):
|
||
params = best_params.get(regime, defaults[regime])
|
||
dist = params.get('distance', defaults[regime]['distance'])
|
||
lines.append(f"#define ML_DCA_DIST_{regime_names[regime]} {float(dist):.1f}")
|
||
lines.append("")
|
||
|
||
lines.append("// ===================================================================")
|
||
lines.append("// LOT MULTIPLIER PER REGIME")
|
||
lines.append("// Conservative (1.0) when volatile, aggressive when calm")
|
||
lines.append("// ===================================================================")
|
||
for regime in range(4):
|
||
params = best_params.get(regime, defaults[regime])
|
||
lm = params.get('lot_multi', defaults[regime]['lot_multi'])
|
||
lines.append(f"#define ML_LOT_MULT_{regime_names[regime]} {float(lm):.2f}")
|
||
lines.append("")
|
||
|
||
lines.append("// ===================================================================")
|
||
lines.append("// MAX DCA ORDERS PER REGIME")
|
||
lines.append("// Fewer DCA in volatile market = limit exposure")
|
||
lines.append("// ===================================================================")
|
||
for regime in range(4):
|
||
params = best_params.get(regime, defaults[regime])
|
||
md = params.get('max_dca', defaults[regime]['max_dca'])
|
||
lines.append(f"#define ML_MAX_DCA_{regime_names[regime]} {int(md)}")
|
||
lines.append("")
|
||
|
||
lines.append("// ===================================================================")
|
||
lines.append("// DANGER ZONE PROTECTION")
|
||
lines.append("// If total lots OR distance exceeds these → STOP DCA immediately")
|
||
lines.append("// These are derived from blow-up pattern analysis")
|
||
lines.append("// ===================================================================")
|
||
lines.append(f"#define ML_MAX_TOTAL_LOTS 5.0 // Max total lots in one chain")
|
||
lines.append(f"#define ML_DANGER_ZONE_DIST 80.0 // Max pips from avg entry → stop DCA")
|
||
lines.append(f"#define ML_DANGER_ZONE_LOTS 3.0 // If lots > this AND dist > 50p → stop")
|
||
lines.append(f"#define ML_EMERGENCY_CUT_DIST 120.0 // Emergency: close ALL if distance > this")
|
||
lines.append(f"#define ML_EMERGENCY_CUT_LOTS 8.0 // Emergency: close ALL if lots > this")
|
||
lines.append("")
|
||
|
||
lines.append("// ===================================================================")
|
||
lines.append("// DISTANCE MULTIPLIER (increases distance for each subsequent DCA)")
|
||
lines.append("// ===================================================================")
|
||
lines.append(f"#define ML_DCA_DIST_MULTIPLIER 1.25 // Each DCA level = prev * 1.25")
|
||
lines.append("")
|
||
|
||
lines.append("// ===================================================================")
|
||
lines.append("// OPTIMIZATION LOG")
|
||
lines.append("// ===================================================================")
|
||
for regime in range(4):
|
||
params = best_params.get(regime, {})
|
||
if params:
|
||
tp = params.get('tp_rate', 0) * 100
|
||
bl = params.get('blowup_rate', 0) * 100
|
||
sc = params.get('score', 0)
|
||
lines.append(f"// {regime_names[regime]}: TP={tp:.1f}% Blowup={bl:.1f}% Score={sc:.1f}")
|
||
|
||
lines.append("")
|
||
lines.append("#endif // ML_PARAMS_MQH")
|
||
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
f.write('\n'.join(lines))
|
||
|
||
print(f"\n[OK] Generated {output_path}")
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def generate_optimization_log(best_params, atr_thresholds, output_path):
|
||
"""Save detailed optimization log as CSV."""
|
||
regime_names = {0: 'LOW_VOL', 1: 'MED_VOL', 2: 'HIGH_VOL', 3: 'EXTREME'}
|
||
|
||
with open(output_path, 'w', newline='', encoding='utf-8') as f:
|
||
writer = csv.writer(f)
|
||
writer.writerow(['timestamp', 'regime', 'distance', 'max_dca', 'lot_multi',
|
||
'tp_rate', 'blowup_rate', 'avg_adverse_pips', 'avg_lots', 'score',
|
||
'atr_threshold'])
|
||
|
||
for regime, params in best_params.items():
|
||
if params:
|
||
writer.writerow([
|
||
datetime.now().isoformat(),
|
||
regime_names[regime],
|
||
params.get('distance', 0),
|
||
params.get('max_dca', 0),
|
||
params.get('lot_multi', 0),
|
||
f"{params.get('tp_rate', 0):.4f}",
|
||
f"{params.get('blowup_rate', 0):.4f}",
|
||
f"{params.get('avg_adverse', 0):.2f}",
|
||
f"{params.get('avg_lots', 0):.2f}",
|
||
f"{params.get('score', 0):.2f}",
|
||
"{:.4f}".format(atr_thresholds.get("atr_" + ["low","med","high","extreme"][regime], 0)),
|
||
])
|
||
|
||
print(f"[OK] Generated {output_path}")
|
||
|
||
|
||
# =============================================================================
|
||
# MAIN
|
||
# =============================================================================
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='DCA ML Optimizer for XAUUSD')
|
||
parser.add_argument('--data', type=str, help='Path to candle CSV data file')
|
||
parser.add_argument('--test', action='store_true', help='Run with sample data')
|
||
parser.add_argument('--export-mt5', action='store_true', help='Also generate MT5 export script')
|
||
parser.add_argument('--lot', type=float, default=0.19, help='Base lot size (default: 0.19)')
|
||
parser.add_argument('--output-dir', type=str, default='.', help='Output directory')
|
||
|
||
args = parser.parse_args()
|
||
|
||
print("=" * 60)
|
||
print(" DCA ML OPTIMIZER v1.0")
|
||
print(" Comarai - AI-Powered Trading Optimization")
|
||
print(" https://comarai.com")
|
||
print("=" * 60)
|
||
|
||
# Load or generate data
|
||
if args.test:
|
||
candles = generate_sample_data(50000)
|
||
elif args.data:
|
||
if not os.path.exists(args.data):
|
||
print(f"[ERROR] File not found: {args.data}")
|
||
sys.exit(1)
|
||
candles = load_candle_data(args.data)
|
||
else:
|
||
print("[INFO] No data file specified. Use --data FILE.csv or --test")
|
||
print(" To export from MT5: File → Save As → CSV (M5 timeframe)")
|
||
sys.exit(0)
|
||
|
||
if len(candles) < 1000:
|
||
print(f"[ERROR] Need at least 1000 candles, got {len(candles)}")
|
||
sys.exit(1)
|
||
|
||
# Feature engineering
|
||
print("\n[STEP 1] Computing features...")
|
||
atr_m5 = compute_atr(candles, period=14)
|
||
features = compute_volatility_features(candles, atr_m5)
|
||
print(f" Computed {len(features)} feature vectors")
|
||
|
||
# ATR thresholds
|
||
print("\n[STEP 2] Computing ATR regime thresholds...")
|
||
atr_thresholds = compute_atr_thresholds(candles, features)
|
||
|
||
# Optimize DCA params
|
||
print("\n[STEP 3] Optimizing DCA parameters per regime...")
|
||
best_params, regime_results = optimize_dca_params(candles, features, lot_size=args.lot)
|
||
|
||
# Danger pattern detection
|
||
print("\n[STEP 4] Detecting danger patterns (blow-up scenarios)...")
|
||
danger_thresholds = detect_danger_patterns(candles, features)
|
||
|
||
# Generate outputs
|
||
print("\n[STEP 5] Generating output files...")
|
||
output_dir = args.output_dir
|
||
|
||
mqh_path = os.path.join(output_dir, 'ml_params.mqh')
|
||
generate_mqh_file(best_params, atr_thresholds, danger_thresholds, mqh_path)
|
||
|
||
log_path = os.path.join(output_dir, 'optimization_log.csv')
|
||
generate_optimization_log(best_params, atr_thresholds, log_path)
|
||
|
||
# Summary
|
||
print("\n" + "=" * 60)
|
||
print(" OPTIMIZATION COMPLETE")
|
||
print("=" * 60)
|
||
print(f" Output: {mqh_path}")
|
||
print(f" Log: {log_path}")
|
||
print(f"\n Next steps:")
|
||
print(f" 1. Review ml_params.mqh (check DCA distances make sense)")
|
||
print(f" 2. Copy ml_params.mqh to MQL5/Include/ folder")
|
||
print(f" 3. Compile IchiDCA_ML_CCBSN.mq5 in MetaEditor")
|
||
print(f" 4. Backtest on MT5 Strategy Tester")
|
||
print("=" * 60)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|