zhutoutoutousan
222 lines
7.5 KiB
Python
222 lines
7.5 KiB
Python
"""
|
|
Data Collection Script for BTCUSD RSI Divergence Training
|
|
Fetches BTCUSD data from MetaTrader 5 and labels it with RSI divergence signals.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
import pandas as pd
|
|
import numpy as np
|
|
import MetaTrader5 as mt5
|
|
from rsi_divergence_detector import RSIDivergenceDetector, DivergenceType
|
|
import pickle
|
|
from tqdm import tqdm
|
|
|
|
|
|
def fetch_mt5_data(symbol: str, timeframe: int, start_date: datetime, end_date: datetime) -> pd.DataFrame:
|
|
"""
|
|
Fetch historical data from MetaTrader 5.
|
|
|
|
Args:
|
|
symbol: Trading symbol (e.g., 'BTCUSD')
|
|
timeframe: MT5 timeframe constant
|
|
start_date: Start date for data
|
|
end_date: End date for data
|
|
|
|
Returns:
|
|
DataFrame with OHLCV data
|
|
"""
|
|
print(f"Fetching {symbol} data from {start_date} to {end_date}...")
|
|
|
|
if not mt5.initialize():
|
|
raise RuntimeError(f"MT5 initialization failed: {mt5.last_error()}")
|
|
|
|
try:
|
|
rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date)
|
|
|
|
if rates is None or len(rates) == 0:
|
|
raise ValueError(f"No data available for {symbol} in the specified date range")
|
|
|
|
df = pd.DataFrame(rates)
|
|
df['time'] = pd.to_datetime(df['time'], unit='s')
|
|
df.set_index('time', inplace=True)
|
|
|
|
# Rename columns to lowercase
|
|
df.columns = [col.lower() for col in df.columns]
|
|
|
|
print(f"Fetched {len(df)} bars")
|
|
return df
|
|
|
|
finally:
|
|
mt5.shutdown()
|
|
|
|
|
|
def prepare_features(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Prepare additional features for training.
|
|
|
|
Args:
|
|
df: DataFrame with OHLCV data
|
|
|
|
Returns:
|
|
DataFrame with additional features
|
|
"""
|
|
feature_df = df.copy()
|
|
|
|
# Price-based features
|
|
feature_df['returns'] = feature_df['close'].pct_change()
|
|
feature_df['high_low_ratio'] = feature_df['high'] / (feature_df['low'] + 1e-10)
|
|
feature_df['close_open_ratio'] = feature_df['close'] / (feature_df['open'] + 1e-10)
|
|
|
|
# Moving averages
|
|
feature_df['sma_20'] = feature_df['close'].rolling(window=20).mean()
|
|
feature_df['sma_50'] = feature_df['close'].rolling(window=50).mean()
|
|
feature_df['ema_20'] = feature_df['close'].ewm(span=20).mean()
|
|
feature_df['ema_50'] = feature_df['close'].ewm(span=50).mean()
|
|
|
|
# ATR
|
|
high_low = feature_df['high'] - feature_df['low']
|
|
high_close = np.abs(feature_df['high'] - feature_df['close'].shift())
|
|
low_close = np.abs(feature_df['low'] - feature_df['close'].shift())
|
|
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
|
|
feature_df['atr'] = tr.rolling(window=14).mean()
|
|
feature_df['atr_pct'] = feature_df['atr'] / (feature_df['close'] + 1e-10)
|
|
|
|
# Volume features
|
|
if 'tick_volume' in feature_df.columns:
|
|
feature_df['volume_ma'] = feature_df['tick_volume'].rolling(window=20).mean()
|
|
feature_df['volume_ratio'] = feature_df['tick_volume'] / (feature_df['volume_ma'] + 1e-10)
|
|
|
|
# Price position relative to range
|
|
feature_df['price_position'] = (feature_df['close'] - feature_df['low'].rolling(20).min()) / (
|
|
feature_df['high'].rolling(20).max() - feature_df['low'].rolling(20).min() + 1e-10
|
|
)
|
|
|
|
return feature_df
|
|
|
|
|
|
def create_sequences(df: pd.DataFrame, lookback: int = 60, prediction_horizon: int = 5) -> tuple:
|
|
"""
|
|
Create sequences for training.
|
|
|
|
Args:
|
|
df: Labeled DataFrame
|
|
lookback: Number of bars to look back
|
|
prediction_horizon: Number of bars ahead to predict
|
|
|
|
Returns:
|
|
Tuple of (X, y) where X is features and y is labels
|
|
"""
|
|
# Feature columns (exclude labels and time-based columns)
|
|
exclude_cols = ['divergence_type', 'divergence_confidence', 'divergence_strength', 'time']
|
|
feature_cols = [col for col in df.columns if col not in exclude_cols]
|
|
|
|
X, y = [], []
|
|
|
|
for i in range(lookback, len(df) - prediction_horizon):
|
|
# Get feature sequence
|
|
X.append(df[feature_cols].iloc[i - lookback:i].values)
|
|
|
|
# Get label (divergence type at current bar)
|
|
y.append(df['divergence_type'].iloc[i])
|
|
|
|
return np.array(X), np.array(y)
|
|
|
|
|
|
def main():
|
|
"""Main function."""
|
|
parser = argparse.ArgumentParser(description='Collect and label BTCUSD data for RSI divergence training')
|
|
parser.add_argument('--symbol', type=str, default='BTCUSD', help='Trading symbol')
|
|
parser.add_argument('--timeframe', type=str, default='H1',
|
|
choices=['M1', 'M5', 'M15', 'M30', 'H1', 'H4', 'D1'],
|
|
help='Timeframe')
|
|
parser.add_argument('--days', type=int, default=365, help='Number of days of historical data')
|
|
parser.add_argument('--rsi-period', type=int, default=14, help='RSI period')
|
|
parser.add_argument('--output', type=str, default='data', help='Output directory')
|
|
parser.add_argument('--min-strength', type=float, default=0.15,
|
|
help='Minimum divergence strength (0-1)')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Convert timeframe string to MT5 constant
|
|
timeframe_map = {
|
|
'M1': mt5.TIMEFRAME_M1,
|
|
'M5': mt5.TIMEFRAME_M5,
|
|
'M15': mt5.TIMEFRAME_M15,
|
|
'M30': mt5.TIMEFRAME_M30,
|
|
'H1': mt5.TIMEFRAME_H1,
|
|
'H4': mt5.TIMEFRAME_H4,
|
|
'D1': mt5.TIMEFRAME_D1
|
|
}
|
|
timeframe = timeframe_map[args.timeframe]
|
|
|
|
# Create output directory
|
|
os.makedirs(args.output, exist_ok=True)
|
|
|
|
# Fetch data
|
|
end_date = datetime.now()
|
|
start_date = end_date - timedelta(days=args.days)
|
|
|
|
print(f"\n{'='*60}")
|
|
print("BTCUSD RSI Divergence Data Collection")
|
|
print(f"{'='*60}\n")
|
|
|
|
df = fetch_mt5_data(args.symbol, timeframe, start_date, end_date)
|
|
|
|
# Prepare features
|
|
print("\nPreparing features...")
|
|
df = prepare_features(df)
|
|
|
|
# Detect and label divergences
|
|
print("\nDetecting RSI divergences...")
|
|
detector = RSIDivergenceDetector(
|
|
rsi_period=args.rsi_period,
|
|
min_divergence_strength=args.min_strength
|
|
)
|
|
|
|
df = detector.label_data(df)
|
|
|
|
# Statistics
|
|
total_bars = len(df)
|
|
labeled_bars = len(df[df['divergence_type'] != DivergenceType.NONE.value])
|
|
|
|
print(f"\n{'='*60}")
|
|
print("Labeling Statistics:")
|
|
print(f"{'='*60}")
|
|
print(f"Total bars: {total_bars}")
|
|
print(f"Bars with divergence: {labeled_bars} ({labeled_bars/total_bars*100:.2f}%)")
|
|
|
|
for div_type in DivergenceType:
|
|
if div_type == DivergenceType.NONE:
|
|
continue
|
|
count = len(df[df['divergence_type'] == div_type.value])
|
|
print(f" {div_type.name}: {count} ({count/total_bars*100:.2f}%)")
|
|
|
|
# Save labeled data
|
|
output_file = os.path.join(args.output, f"{args.symbol}_{args.timeframe}_labeled.csv")
|
|
df.to_csv(output_file)
|
|
print(f"\nLabeled data saved to: {output_file}")
|
|
|
|
# Save detector parameters
|
|
detector_params = {
|
|
'rsi_period': args.rsi_period,
|
|
'min_swing_bars': detector.min_swing_bars,
|
|
'max_swing_bars': detector.max_swing_bars,
|
|
'min_divergence_strength': args.min_strength
|
|
}
|
|
|
|
params_file = os.path.join(args.output, f"{args.symbol}_{args.timeframe}_detector_params.pkl")
|
|
with open(params_file, 'wb') as f:
|
|
pickle.dump(detector_params, f)
|
|
print(f"Detector parameters saved to: {params_file}")
|
|
|
|
print(f"\n{'='*60}")
|
|
print("Data collection completed!")
|
|
print(f"{'='*60}\n")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|