commit 559d78388d30aa62c783cfdb7cfec0d9c95629d9 Author: songkunling Date: Fri Jul 25 17:35:01 2025 +0800 add files diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..37bad80 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +/GEMINI.md +/logs +/myenv +__pycache__/ \ No newline at end of file diff --git a/backtest.py b/backtest.py new file mode 100644 index 0000000..ce1b370 --- /dev/null +++ b/backtest.py @@ -0,0 +1,45 @@ +import pandas as pd + +class BacktestEngine: + def __init__(self, df): + """ + df: 包含历史k线的DataFrame，至少包括open, high, low, close字段 + """ + self.df = df + + def run_strategy(self, strategy): + """ + 执行策略的run_backtest，得到信号序列 + """ + return strategy.run_backtest(self.df) + + def combine_signals(self, signals_list, weights, buy_threshold, sell_threshold): + """ + 多策略信号加权合成，并根据阈值生成最终信号 + 返回合成信号序列 + """ + df_signals = pd.concat(signals_list, axis=1).fillna(0) + weighted_signals = df_signals * weights + combined = weighted_signals.sum(axis=1) + + def apply_threshold(score): + if score >= buy_threshold: + return 1 + elif score <= sell_threshold: + return -1 + else: + return 0 + + combined_signal = combined.apply(apply_threshold) + return combined_signal + + def calc_returns(self, signals): + """ + 根据信号计算策略回测收益率（简化版） + """ + df = self.df.copy() + df['signal'] = signals.shift(1).fillna(0) # 防止未来函数 + df['returns'] = df['close'].pct_change() + df['strategy_returns'] = df['signal'] * df['returns'] + cum_ret = (1 + df['strategy_returns']).cumprod() - 1 + return cum_ret diff --git a/config.py b/config.py new file mode 100644 index 0000000..a2513f4 --- /dev/null +++ b/config.py @@ -0,0 +1,24 @@ +from strategies import ma_cross, rsi, bollinger, mean_reversion, momentum_breakout, macd, kdj, turtle, daily_breakout, profit_protect, resilient_trend + +SYMBOL = "XAUUSD" +INTERVAL = 60 # 秒 +INITIAL_CAPITAL = 10000 # 初始资金 + +# 策略权重配置 (策略实例, 权重) +STRATEGIES = [ + (ma_cross.Strategy(), 2.28), + (rsi.Strategy(), 1.84), + (bollinger.Strategy(), 1.46), + (mean_reversion.Strategy(), 1.90), + (momentum_breakout.Strategy(), 2.77), + (macd.Strategy(), 1.70), + (kdj.Strategy(), 0.65), + (turtle.Strategy(), 0.68), + (daily_breakout.Strategy(), 0.66), + (profit_protect.Strategy(), 0.57), + (resilient_trend.Strategy(), 1.0), # 新增带容错的趋势策略 +] + +# 交易信号阈值 +BUY_THRESHOLD = 1.5 +SELL_THRESHOLD = -1.5 diff --git a/logger.py b/logger.py new file mode 100644 index 0000000..cc24419 --- /dev/null +++ b/logger.py @@ -0,0 +1,40 @@ +import logging +import os + +# 全局变量，用于存储logger实例 +_logger_instance = None + +def setup_logger(): + global _logger_instance + if _logger_instance: + return _logger_instance + + log_dir = "logs" + os.makedirs(log_dir, exist_ok=True) + log_file = os.path.join(log_dir, "strategy.log") + + logger = logging.getLogger("StrategyLogger") + logger.setLevel(logging.DEBUG) + + # 防止重复添加handler + if not logger.handlers: + # 输出到控制台 + console_handler = logging.StreamHandler() + console_handler.setLevel(logging.INFO) + + # 输出到文件 + file_handler = logging.FileHandler(log_file, encoding="utf-8") + file_handler.setLevel(logging.DEBUG) + + formatter = logging.Formatter('%(asctime)s - [%(levelname)s] - %(message)s') + console_handler.setFormatter(formatter) + file_handler.setFormatter(formatter) + + logger.addHandler(console_handler) + logger.addHandler(file_handler) + + _logger_instance = logger + return logger + +# 直接导出一个已经初始化好的logger实例 +logger = setup_logger() diff --git a/main.py b/main.py new file mode 100644 index 0000000..8e5719c --- /dev/null +++ b/main.py @@ -0,0 +1,90 @@ +import importlib +import pandas as pd +from utils import initialize, shutdown, get_rates, close_all, send_order +from backtest import BacktestEngine +from logger import logger +from config import INITIAL_CAPITAL, STRATEGIES, BUY_THRESHOLD, SELL_THRESHOLD + +# 导入优化器 +from optimizer import run_optimizer + +def run_realtime(): + if not initialize(): + logger.error("MT5初始化失败") + return + + signals = [] + weights = [] + for strat, weight in STRATEGIES: + try: + logger.info(f"执行策略：{strat.__class__.__module__}") + signal = strat.generate_signal() + signals.append(signal) + weights.append(weight) + except Exception as e: + logger.exception(f"运行策略 {strat.__class__.__module__} 时出错：{e}") + + weighted_signal_sum = sum(s * w for s, w in zip(signals, weights)) + + if weighted_signal_sum >= BUY_THRESHOLD: + logger.info(f"加权信号总和 ({weighted_signal_sum:.2f}) 达到买入阈值 ({BUY_THRESHOLD})，发送买入信号") + close_all("XAUUSD") + send_order("XAUUSD", 'buy') + elif weighted_signal_sum <= SELL_THRESHOLD: + logger.info(f"加权信号总和 ({weighted_signal_sum:.2f}) 达到卖出阈值 ({SELL_THRESHOLD})，发送卖出信号") + close_all("XAUUSD") + send_order("XAUUSD", 'sell') + else: + logger.info(f"加权信号总和 ({weighted_signal_sum:.2f}) 未达到交易阈值，无操作") + + shutdown() + +def run_backtest(): + if not initialize(): + logger.error("MT5初始化失败") + return + + symbol = "XAUUSD" + timeframe = 1 # M1 + count = 50000 + rates = get_rates(symbol, timeframe, count) + shutdown() + + if rates is None: + logger.error("获取历史数据失败") + return + + logger.info(f"初始资金: {INITIAL_CAPITAL}") + + df = pd.DataFrame(rates) + engine = BacktestEngine(df) + signals_list = [] + weights = [] + + for strat, weight in STRATEGIES: + try: + logger.info(f"回测策略：{strat.__class__.__module__}") + signals = engine.run_strategy(strat) + signals_list.append(signals) + weights.append(weight) + except Exception as e: + logger.exception(f"回测策略 {strat.__class__.__module__} 时出错：{e}") + + combined_signal = engine.combine_signals(signals_list, weights, BUY_THRESHOLD, SELL_THRESHOLD) + cum_ret = engine.calc_returns(combined_signal) + final_capital = INITIAL_CAPITAL * (1 + cum_ret.iloc[-1]) + logger.info("策略组合回测完成") + logger.info(f"最终资金: {final_capital:.2f}") + logger.info(cum_ret.tail()) + + +if __name__ == "__main__": + # --- 选择运行模式 --- + # 1. 运行一次回测 (使用config.py中的默认权重) + # run_backtest() + + # 2. 运行实盘交易 (使用config.py中的默认权重) + # run_realtime() + + # 3. 运行遗传算法优化，寻找最佳权重 + run_optimizer() diff --git a/optimizer.py b/optimizer.py new file mode 100644 index 0000000..bf8f0aa --- /dev/null +++ b/optimizer.py @@ -0,0 +1,115 @@ + +import random +import numpy as np +from deap import base, creator, tools, algorithms +import multiprocessing + +import pandas as pd +from utils import initialize, shutdown, get_rates +from backtest import BacktestEngine +from logger import logger +from config import INITIAL_CAPITAL, STRATEGIES, BUY_THRESHOLD, SELL_THRESHOLD + +# 1. 定义适应度函数 (已优化) +def evaluate_fitness(individual, df_data): + """ + 评估函数现在接收预先加载的DataFrame作为参数，避免了重复IO。 + 输入: + - individual: 一个代表策略权重的列表。 + - df_data: 包含历史K线数据的Pandas DataFrame。 + 输出: 一个元组，包含适应度分数（最终资金）。 + """ + weights = individual + + # 直接使用传入的df_data，不再需要get_rates + engine = BacktestEngine(df_data) + signals_list = [] + + strategy_instances = [s for s, w in STRATEGIES] + + for strat in strategy_instances: + signals = engine.run_strategy(strat) + signals_list.append(signals) + + combined_signal = engine.combine_signals(signals_list, weights, BUY_THRESHOLD, SELL_THRESHOLD) + cum_ret = engine.calc_returns(combined_signal) + final_capital = INITIAL_CAPITAL * (1 + cum_ret.iloc[-1]) + + # 在优化过程中，可以注释掉这行日志以提高速度，因为它会大量输出 + # logger.info(f"评估权重: {[f'{w:.2f}' for w in weights]} -> 最终资金: {final_capital:.2f}") + + return (final_capital,) + +# 2. 设置遗传算法 (已优化) +def run_optimizer(): + """ + 配置并运行遗传算法 + """ + # --- 数据预加载 --- + logger.info("--- 开始遗传算法优化 --- ") + logger.info("步骤 1/4: 初始化MT5并预加载历史数据...") + if not initialize(): + logger.error("MT5初始化失败，无法开始优化") + return + + symbol = "XAUUSD" + timeframe = 1 # M1 + count = 50000 # 使用与回测相同的数据量 + rates = get_rates(symbol, timeframe, count) + shutdown() # 获取数据后即可关闭连接 + + if rates is None: + logger.error("获取历史数据失败，优化终止") + return + + df_historical_data = pd.DataFrame(rates) + logger.info(f"历史数据加载完成，共 {len(df_historical_data)} 条记录。") + + # --- DEAP 设置 --- + logger.info("步骤 2/4: 配置遗传算法...") + creator.create("FitnessMax", base.Fitness, weights=(1.0,)) + creator.create("Individual", list, fitness=creator.FitnessMax) + + toolbox = base.Toolbox() + toolbox.register("attr_float", random.uniform, 0.1, 2.0) + num_strategies = len(STRATEGIES) + toolbox.register("individual", tools.initRepeat, creator.Individual, toolbox.attr_float, n=num_strategies) + toolbox.register("population", tools.initRepeat, list, toolbox.individual) + + toolbox.register("evaluate", evaluate_fitness, df_data=df_historical_data) + toolbox.register("mate", tools.cxTwoPoint) + toolbox.register("mutate", tools.mutGaussian, mu=0, sigma=0.5, indpb=0.2) + toolbox.register("select", tools.selTournament, tournsize=3) + + # --- 并行计算设置 --- + logger.info("步骤 3/4: 配置并行计算和统计...") + pool = multiprocessing.Pool() + toolbox.register("map", pool.map) + + # --- 统计功能设置 --- + stats = tools.Statistics(lambda ind: ind.fitness.values) + stats.register("avg", np.mean) + stats.register("std", np.std) + stats.register("min", np.min) + stats.register("max", np.max) + + # --- 运行算法 --- + population = toolbox.population(n=50) + ngen = 20 + cxpb = 0.5 + mutpb = 0.2 + + logger.info(f"步骤 4/4: 开始并行进化... (种群大小: {len(population)}, 进化代数: {ngen})") + algorithms.eaSimple(population, toolbox, cxpb, mutpb, ngen, stats=stats, verbose=True) + + pool.close() + + # --- 结果 --- + best_individual = tools.selBest(population, k=1)[0] + best_fitness = best_individual.fitness.values[0] + + logger.info("--- 遗传算法优化结束 ---") + logger.info(f"找到的最佳权重: {[f'{w:.2f}' for w in best_individual]}") + logger.info(f"对应的最佳最终资金: {best_fitness:.2f}") + + return best_individual, best_fitness diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f2676cf --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +MetaTrader5 +pandas +deap diff --git a/strategies/bollinger.py b/strategies/bollinger.py new file mode 100644 index 0000000..f949e56 --- /dev/null +++ b/strategies/bollinger.py @@ -0,0 +1,57 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, close_all, send_order +from logger import logger + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.bollinger_period = 20 + self.bollinger_std_dev = 2 + + def _calculate_indicators(self, df): + """ + 计算布林带指标 + """ + mean = df['close'].rolling(self.bollinger_period).mean() + std = df['close'].rolling(self.bollinger_period).std() + df['upper_band'] = mean + self.bollinger_std_dev * std + df['lower_band'] = mean - self.bollinger_std_dev * std + return df + + def generate_signal(self): + """ + 布林带策略实盘： + 当价格跌破下轨买入，涨破上轨卖出。 + """ + rates = get_rates(self.symbol, self.timeframe, self.bollinger_period + 30) + if rates is None or len(rates) < self.bollinger_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['close'].iloc[-2] < df['lower_band'].iloc[-2]: + logger.info(f"价格跌破下轨，产生买入信号: {self.symbol}") + return 1 + elif df['close'].iloc[-2] > df['upper_band'].iloc[-2]: + logger.info(f"价格涨破上轨，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + 布林带策略回测： + 价格突破下轨买入，突破上轨卖出 + 返回信号序列：1买入，-1卖出，0无操作 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + for i in range(self.bollinger_period, len(df)): + if df['close'].iloc[i-1] < df['lower_band'].iloc[i-1]: + signals.iat[i] = 1 + elif df['close'].iloc[i-1] > df['upper_band'].iloc[i-1]: + signals.iat[i] = -1 + return signals diff --git a/strategies/daily_breakout.py b/strategies/daily_breakout.py new file mode 100644 index 0000000..62a4e06 --- /dev/null +++ b/strategies/daily_breakout.py @@ -0,0 +1,68 @@ +import MetaTrader5 as mt5 +import pandas as pd +from datetime import datetime +from utils import get_rates, has_open_position, close_all, send_order +from logger import logger + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + + def _calculate_indicators(self, df): + """ + 计算日内突破指标 + """ + df['time'] = pd.to_datetime(df['time'], unit='s') + today = datetime.now().date() + day_data = df[df['time'].dt.date == today] + if day_data.empty: + return df, None, None + day_high = day_data['high'].max() + day_low = day_data['low'].min() + return df, day_high, day_low + + def generate_signal(self): + """ + 日内突破策略实盘 + 当价格突破当日最高买入，突破当日最低卖出 + """ + rates = get_rates(self.symbol, self.timeframe, 1440) # 24 hours * 60 minutes + if rates is None or len(rates) < 2: + return 0 + df = pd.DataFrame(rates) + df, day_high, day_low = self._calculate_indicators(df) + + if day_high is None or day_low is None: + return 0 + + if df['close'].iloc[-2] > day_high: + logger.info(f"价格突破当日最高，产生买入信号: {self.symbol}") + return 1 + elif df['close'].iloc[-2] < day_low: + logger.info(f"价格突破当日最低，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + 日内突破回测方法 + 计算每个交易日的高低点，突破买卖信号 + """ + df = df.copy() + df['time'] = pd.to_datetime(df['time'], unit='s') + + signals = pd.Series(0, index=df.index) + + grouped = df.groupby(df['time'].dt.date) + + for date, group in grouped: + day_high = group['high'].max() + day_low = group['low'].min() + for i, row in group.iterrows(): + if row['close'] > day_high: + signals.loc[i] = 1 + elif row['close'] < day_low: + signals.loc[i] = -1 + + return signals diff --git a/strategies/kdj.py b/strategies/kdj.py new file mode 100644 index 0000000..323906d --- /dev/null +++ b/strategies/kdj.py @@ -0,0 +1,60 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, has_open_position, close_all, send_order +from logger import logger + + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.kdj_period = 9 + self.kdj_buy_threshold = 10 + self.kdj_sell_threshold = 90 + + def _calculate_indicators(self, df): + """ + 计算KDJ指标 + """ + low_min = df['low'].rolling(self.kdj_period).min() + high_max = df['high'].rolling(self.kdj_period).max() + rsv = (df['close'] - low_min) / (high_max - low_min) * 100 + df['k'] = rsv.ewm(com=2).mean() + df['d'] = df['k'].ewm(com=2).mean() + df['j'] = 3 * df['k'] - 2 * df['d'] + return df + + def generate_signal(self): + """ + KDJ策略实盘 + J值小于10买入，大于90卖出 + """ + rates = get_rates(self.symbol, self.timeframe, self.kdj_period + 30) + if rates is None or len(rates) < self.kdj_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['j'].iloc[-2] < self.kdj_buy_threshold: + logger.info(f"J值小于{self.kdj_buy_threshold}，产生买入信号: {self.symbol}") + return 1 + elif df['j'].iloc[-2] > self.kdj_sell_threshold: + logger.info(f"J值大于{self.kdj_sell_threshold}，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + KDJ回测方法 + 根据J值极端生成信号 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + for i in range(self.kdj_period, len(df)): + if df['j'].iloc[i-1] < self.kdj_buy_threshold: + signals.iat[i] = 1 + elif df['j'].iloc[i-1] > self.kdj_sell_threshold: + signals.iat[i] = -1 + return signals diff --git a/strategies/ma_cross.py b/strategies/ma_cross.py new file mode 100644 index 0000000..16a0236 --- /dev/null +++ b/strategies/ma_cross.py @@ -0,0 +1,56 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, close_all, send_order +from logger import logger + + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.fast_ma_period = 5 + self.slow_ma_period = 20 + + def _calculate_indicators(self, df): + """ + 计算技术指标 + """ + df['fast_ma'] = df['close'].rolling(self.fast_ma_period).mean() + df['slow_ma'] = df['close'].rolling(self.slow_ma_period).mean() + return df + + def generate_signal(self): + """ + 均线交叉策略实盘： + 短期均线上穿长期均线买入，下穿卖出。 + """ + rates = get_rates(self.symbol, self.timeframe, self.slow_ma_period + 30) + if rates is None or len(rates) < self.slow_ma_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['fast_ma'].iloc[-2] > df['slow_ma'].iloc[-2] and df['fast_ma'].iloc[-3] <= df['slow_ma'].iloc[-3]: + logger.info(f"短期均线上穿长期均线，产生买入信号: {self.symbol}") + return 1 + elif df['fast_ma'].iloc[-2] < df['slow_ma'].iloc[-2] and df['fast_ma'].iloc[-3] >= df['slow_ma'].iloc[-3]: + logger.info(f"短期均线下穿长期均线，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + 均线交叉回测： + 短期均线和长期均线交叉产生信号 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + # 从 slow_ma_period 开始循环，避免早期数据 NaN 问题 + for i in range(self.slow_ma_period, len(df)): + if df['fast_ma'].iloc[i-1] > df['slow_ma'].iloc[i-1] and df['fast_ma'].iloc[i-2] <= df['slow_ma'].iloc[i-2]: + signals.iat[i] = 1 + elif df['fast_ma'].iloc[i-1] < df['slow_ma'].iloc[i-1] and df['fast_ma'].iloc[i-2] >= df['slow_ma'].iloc[i-2]: + signals.iat[i] = -1 + return signals diff --git a/strategies/macd.py b/strategies/macd.py new file mode 100644 index 0000000..31e2943 --- /dev/null +++ b/strategies/macd.py @@ -0,0 +1,58 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, has_open_position, close_all, send_order +from logger import logger + + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.fast_ema_period = 12 + self.slow_ema_period = 26 + self.signal_period = 9 + + def _calculate_indicators(self, df): + """ + 计算MACD指标 + """ + df['exp12'] = df['close'].ewm(span=self.fast_ema_period, adjust=False).mean() + df['exp26'] = df['close'].ewm(span=self.slow_ema_period, adjust=False).mean() + df['dif'] = df['exp12'] - df['exp26'] + df['dea'] = df['dif'].ewm(span=self.signal_period, adjust=False).mean() + return df + + def generate_signal(self): + """ + MACD策略实盘 + DIF线上穿DEA买入，反之卖出 + """ + rates = get_rates(self.symbol, self.timeframe, self.slow_ema_period + self.signal_period + 30) + if rates is None or len(rates) < self.slow_ema_period + self.signal_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['dif'].iloc[-2] > df['dea'].iloc[-2] and df['dif'].iloc[-3] <= df['dea'].iloc[-3]: + logger.info(f"DIF线上穿DEA，产生买入信号: {self.symbol}") + return 1 + elif df['dif'].iloc[-2] < df['dea'].iloc[-2] and df['dif'].iloc[-3] >= df['dea'].iloc[-3]: + logger.info(f"DIF线下穿DEA，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + MACD回测方法 + 根据DIF和DEA金叉死叉生成信号 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + for i in range(2, len(df)): + if df['dif'].iloc[i-1] > df['dea'].iloc[i-1] and df['dif'].iloc[i-2] <= df['dea'].iloc[i-2]: + signals.iat[i] = 1 + elif df['dif'].iloc[i-1] < df['dea'].iloc[i-1] and df['dif'].iloc[i-2] >= df['dea'].iloc[i-2]: + signals.iat[i] = -1 + return signals diff --git a/strategies/mean_reversion.py b/strategies/mean_reversion.py new file mode 100644 index 0000000..909b23b --- /dev/null +++ b/strategies/mean_reversion.py @@ -0,0 +1,57 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, close_all, send_order +from logger import logger + + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.mean_reversion_period = 20 + self.mean_reversion_std_dev = 2 + + def _calculate_indicators(self, df): + """ + 计算均值回归指标 + """ + mean = df['close'].rolling(self.mean_reversion_period).mean() + std = df['close'].rolling(self.mean_reversion_period).std() + df['upper_band'] = mean + self.mean_reversion_std_dev * std + df['lower_band'] = mean - self.mean_reversion_std_dev * std + return df + + def generate_signal(self): + """ + 均值回归策略实盘： + 当价格超过20日均线正负2个标准差买卖。 + """ + rates = get_rates(self.symbol, self.timeframe, self.mean_reversion_period + 30) + if rates is None or len(rates) < self.mean_reversion_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['close'].iloc[-2] > df['upper_band'].iloc[-2]: + logger.info(f"价格超过上轨，产生卖出信号: {self.symbol}") + return -1 + elif df['close'].iloc[-2] < df['lower_band'].iloc[-2]: + logger.info(f"价格低于下轨，产生买入信号: {self.symbol}") + return 1 + return 0 + + def run_backtest(self, df): + """ + 均值回归回测： + 价格突破上下轨卖出/买入 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + for i in range(self.mean_reversion_period, len(df)): + if df['close'].iloc[i-1] > df['upper_band'].iloc[i-1]: + signals.iat[i] = -1 + elif df['close'].iloc[i-1] < df['lower_band'].iloc[i-1]: + signals.iat[i] = 1 + return signals diff --git a/strategies/momentum_breakout.py b/strategies/momentum_breakout.py new file mode 100644 index 0000000..91a6cc0 --- /dev/null +++ b/strategies/momentum_breakout.py @@ -0,0 +1,54 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, has_open_position, close_all, send_order +from logger import logger + + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.breakout_period = 20 + + def _calculate_indicators(self, df): + """ + 计算动量突破指标 + """ + df['high_20'] = df['high'].rolling(self.breakout_period).max() + df['low_20'] = df['low'].rolling(self.breakout_period).min() + return df + + def generate_signal(self): + """ + 动量突破策略实盘 + 价格突破过去20根K线最高点买入，突破最低点卖出 + """ + rates = get_rates(self.symbol, self.timeframe, self.breakout_period + 30) + if rates is None or len(rates) < self.breakout_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['close'].iloc[-2] > df['high_20'].iloc[-3]: + logger.info(f"价格突破{self.breakout_period}日最高点，产生买入信号: {self.symbol}") + return 1 + elif df['close'].iloc[-2] < df['low_20'].iloc[-3]: + logger.info(f"价格突破{self.breakout_period}日最低点，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + 动量突破回测方法 + 过去20根K线最高最低突破生成买卖信号 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + for i in range(self.breakout_period, len(df)): + if df['close'].iloc[i-1] > df['high_20'].iloc[i-2]: + signals.iat[i] = 1 + elif df['close'].iloc[i-1] < df['low_20'].iloc[i-2]: + signals.iat[i] = -1 + return signals diff --git a/strategies/profit_protect.py b/strategies/profit_protect.py new file mode 100644 index 0000000..aafba8e --- /dev/null +++ b/strategies/profit_protect.py @@ -0,0 +1,79 @@ + +import pandas as pd +from logger import logger + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + # --- 策略核心参数 --- + # 固定止损线：亏损10%则卖出 + self.stop_loss_pct = -0.10 + # 利润回撤百分比：从最高利润点回撤30%则卖出 + self.profit_retracement_pct = 0.30 + # 追踪止损的激活阈值：当利润超过5%后，才开始启动追踪止损逻辑 + self.min_profit_for_trailing = 0.05 + + def generate_signal(self): + """ + 此策略为资金管理和退出策略，不产生独立的买入信号。 + 实盘逻辑应与其他策略结合，此处仅为框架完整性。 + """ + logger.warning("ProfitProtect策略是一个退出策略，不应单独用于实盘产生信号。") + return 0 + + def run_backtest(self, df): + """ + 盈利保护策略回测： + - 固定止损：亏损10%卖出。 + - 追踪止损：利润超过5%后启动，从最高利润点回撤30%卖出。 + 为了独立回测，本策略会在一开始买入，然后应用退出逻辑。 + """ + df = df.copy() + signals = pd.Series(0, index=df.index) + + if len(df) < 2: + return signals + + # --- 回测状态变量 --- + position_open = False + entry_price = 0.0 + peak_profit_pct = 0.0 # 记录达到的最高利润百分比 + + for i in range(len(df)): + # 如果没有持仓，就在第一个机会买入（用于独立回测） + if not position_open: + position_open = True + entry_price = df['close'].iloc[i] + signals.iat[i] = 1 # 买入信号 + peak_profit_pct = 0.0 # 重置最高利润 + continue + + # 如果有持仓，则执行退出逻辑 + if position_open: + current_price = df['close'].iloc[i] + current_profit_pct = (current_price - entry_price) / entry_price + + # 1. 更新最高利润点 + peak_profit_pct = max(peak_profit_pct, current_profit_pct) + + # 2. 检查固定止损条件 + if current_profit_pct <= self.stop_loss_pct: + logger.info(f"索引 {i}: 触发固定止损。入场价: {entry_price:.2f}, 当前价: {current_price:.2f}, 亏损: {current_profit_pct:.2%}") + signals.iat[i] = -1 # 卖出信号 + position_open = False # 平仓 + continue + + # 3. 检查追踪止损条件 + # 只有当最高利润超过了激活阈值，才开始计算回撤 + if peak_profit_pct > self.min_profit_for_trailing: + retracement_from_peak = (peak_profit_pct - current_profit_pct) + + # 避免除以零或负数的情况 + if peak_profit_pct > 0: + retracement_pct = retracement_from_peak / peak_profit_pct + if retracement_pct >= self.profit_retracement_pct: + logger.info(f"索引 {i}: 触发追踪止损。最高利润: {peak_profit_pct:.2%}, 当前利润: {current_profit_pct:.2%}, 回撤超过30%") + signals.iat[i] = -1 # 卖出信号 + position_open = False # 平仓 + continue + return signals diff --git a/strategies/resilient_trend.py b/strategies/resilient_trend.py new file mode 100644 index 0000000..9152ce1 --- /dev/null +++ b/strategies/resilient_trend.py @@ -0,0 +1,113 @@ + +import pandas as pd +from logger import logger +from utils import get_rates + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + # --- 策略核心参数 --- + self.trend_period = 50 + self.retracement_tolerance = 0.30 + + # --- 策略状态变量 --- + self.current_trend = "none" # none, uptrend, downtrend + self.trend_peak = 0.0 # 上升趋势中的最高价 + self.trend_trough = float('inf') # 下降趋势中的最低价 + + def generate_signal(self): + """ + 带状态维护的实盘信号生成方法。 + """ + # 获取足够的数据来计算滚动高低点 + rates = get_rates(self.symbol, 1, self.trend_period + 5) + if rates is None or len(rates) < self.trend_period: + return 0 # 数据不足，不产生信号 + + df = pd.DataFrame(rates) + + # 获取当前价格和用于判断突破的历史高低点 + current_price = df['close'].iloc[-1] + high_period = df['high'].rolling(self.trend_period).max().iloc[-2] + low_period = df['low'].rolling(self.trend_period).min().iloc[-2] + + signal = 0 + + # 状态 1: 当前无趋势，等待趋势开始 + if self.current_trend == "none": + if current_price > high_period: + self.current_trend = "uptrend" + self.trend_peak = current_price + signal = 1 + logger.info(f"实盘: 突破进入上升趋势，买入价: {current_price:.2f}") + elif current_price < low_period: + self.current_trend = "downtrend" + self.trend_trough = current_price + signal = -1 + logger.info(f"实盘: 跌破进入下降趋势，卖出价: {current_price:.2f}") + + # 状态 2: 当前处于上升趋势 + elif self.current_trend == "uptrend": + if current_price < self.trend_peak * (1 - self.retracement_tolerance): + logger.info(f"实盘: 上升趋势结束。最高点: {self.trend_peak:.2f}, 当前价: {current_price:.2f}。平仓卖出。") + signal = -1 + self.current_trend = "none" # 重置状态 + else: + self.trend_peak = max(self.trend_peak, current_price) + + # 状态 3: 当前处于下降趋势 + elif self.current_trend == "downtrend": + if current_price > self.trend_trough * (1 + self.retracement_tolerance): + logger.info(f"实盘: 下降趋势结束。最低点: {self.trend_trough:.2f}, 当前价: {current_price:.2f}。平仓买入。") + signal = 1 + self.current_trend = "none" # 重置状态 + else: + self.trend_trough = min(self.trend_trough, current_price) + + return signal + + def run_backtest(self, df): + """ + 带容错的趋势跟踪策略回测： + - 突破N周期高点，进入上升趋势，回撤30%则趋势结束。 + - 跌破N周期低点，进入下降趋势，反弹30%则趋势结束。 + """ + df = df.copy() + signals = pd.Series(0, index=df.index) + + df['high_period'] = df['high'].rolling(self.trend_period).max().shift(1) + df['low_period'] = df['low'].rolling(self.trend_period).min().shift(1) + + # 回测时使用局部变量来管理状态，避免干扰实盘状态 + backtest_trend = "none" + backtest_peak = 0.0 + backtest_trough = float('inf') + + for i in range(self.trend_period, len(df)): + current_price = df['close'].iloc[i] + + if backtest_trend == "none": + if current_price > df['high_period'].iloc[i]: + backtest_trend = "uptrend" + backtest_peak = current_price + signals.iat[i] = 1 + elif current_price < df['low_period'].iloc[i]: + backtest_trend = "downtrend" + backtest_trough = current_price + signals.iat[i] = -1 + + elif backtest_trend == "uptrend": + if current_price < backtest_peak * (1 - self.retracement_tolerance): + signals.iat[i] = -1 + backtest_trend = "none" + else: + backtest_peak = max(backtest_peak, current_price) + + elif backtest_trend == "downtrend": + if current_price > backtest_trough * (1 + self.retracement_tolerance): + signals.iat[i] = 1 + backtest_trend = "none" + else: + backtest_trough = min(backtest_trough, current_price) + + return signals diff --git a/strategies/rsi.py b/strategies/rsi.py new file mode 100644 index 0000000..469fe54 --- /dev/null +++ b/strategies/rsi.py @@ -0,0 +1,59 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, close_all, send_order +from logger import logger + + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.rsi_period = 14 + self.rsi_buy_threshold = 30 + self.rsi_sell_threshold = 70 + + def _calculate_indicators(self, df): + """ + 计算RSI指标 + """ + delta = df['close'].diff() + gain = delta.where(delta > 0, 0).rolling(self.rsi_period).mean() + loss = -delta.where(delta < 0, 0).rolling(self.rsi_period).mean() + rs = gain / loss + df['rsi'] = 100 - (100 / (1 + rs)) + return df + + def generate_signal(self): + """ + RSI策略实盘： + RSI < 30买入，RSI > 70卖出。 + """ + rates = get_rates(self.symbol, self.timeframe, self.rsi_period + 30) + if rates is None or len(rates) < self.rsi_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['rsi'].iloc[-2] < self.rsi_buy_threshold: + logger.info(f"RSI小于{self.rsi_buy_threshold}，产生买入信号: {self.symbol}") + return 1 + elif df['rsi'].iloc[-2] > self.rsi_sell_threshold: + logger.info(f"RSI大于{self.rsi_sell_threshold}，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + RSI回测： + RSI < 30买入，RSI > 70卖出。 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + for i in range(self.rsi_period, len(df)): + if df['rsi'].iloc[i-1] < self.rsi_buy_threshold: + signals.iat[i] = 1 + elif df['rsi'].iloc[i-1] > self.rsi_sell_threshold: + signals.iat[i] = -1 + return signals diff --git a/strategies/turtle.py b/strategies/turtle.py new file mode 100644 index 0000000..827b3b7 --- /dev/null +++ b/strategies/turtle.py @@ -0,0 +1,54 @@ +import MetaTrader5 as mt5 +import pandas as pd +from utils import get_rates, has_open_position, close_all, send_order +from logger import logger + + +class Strategy: + def __init__(self): + self.symbol = "XAUUSD" + self.timeframe = mt5.TIMEFRAME_M1 + self.turtle_period = 20 + + def _calculate_indicators(self, df): + """ + 计算海龟交易指标 + """ + df['high_20'] = df['high'].rolling(self.turtle_period).max() + df['low_20'] = df['low'].rolling(self.turtle_period).min() + return df + + def generate_signal(self): + """ + 海龟交易策略实盘 + 20日最高突破买入，20日最低突破卖出 + """ + rates = get_rates(self.symbol, self.timeframe, self.turtle_period + 30) + if rates is None or len(rates) < self.turtle_period: + return 0 + df = pd.DataFrame(rates) + df = self._calculate_indicators(df) + + if df['close'].iloc[-2] > df['high_20'].iloc[-3]: + logger.info(f"价格突破{self.turtle_period}日最高点，产生买入信号: {self.symbol}") + return 1 + elif df['close'].iloc[-2] < df['low_20'].iloc[-3]: + logger.info(f"价格突破{self.turtle_period}日最低点，产生卖出信号: {self.symbol}") + return -1 + return 0 + + def run_backtest(self, df): + """ + 海龟交易回测方法 + 根据20日高低突破生成买卖信号 + """ + df = df.copy() + df = self._calculate_indicators(df) + + signals = pd.Series(0, index=df.index) + for i in range(self.turtle_period, len(df)): + if df['close'].iloc[i-1] > df['high_20'].iloc[i-2]: + signals.iat[i] = 1 + elif df['close'].iloc[i-1] < df['low_20'].iloc[i-2]: + signals.iat[i] = -1 + return signals diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..7d2d4e3 --- /dev/null +++ b/utils.py @@ -0,0 +1,70 @@ +import MetaTrader5 as mt5 +import pandas as pd +from logger import setup_logger +logger = setup_logger() + +def initialize(): + if not mt5.initialize(): + logger.error("MT5初始化失败，错误代码：%d", mt5.last_error()) + return False + return True + +def shutdown(): + mt5.shutdown() + +def get_rates(symbol, timeframe, count): + rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, count) + if rates is None: + logger.info(f"获取{symbol}历史数据失败") + return None + return rates + +def has_open_position(symbol): + positions = mt5.positions_get(symbol=symbol) + return positions is not None and len(positions) > 0 + +def close_all(symbol): + positions = mt5.positions_get(symbol=symbol) + if positions is None: + return + for pos in positions: + request = { + "action": mt5.TRADE_ACTION_DEAL, + "position": pos.ticket, + "symbol": symbol, + "volume": pos.volume, + "type": mt5.ORDER_TYPE_SELL if pos.type == 0 else mt5.ORDER_TYPE_BUY, + "price": mt5.symbol_info_tick(symbol).bid if pos.type == 0 else mt5.symbol_info_tick(symbol).ask, + "deviation": 20, + "magic": 234000, + "comment": "Close position", + "type_filling": mt5.ORDER_FILLING_RETURN, + } + mt5.order_send(request) + +def send_order(symbol, order_type, volume=0.01): + symbol_info_tick = mt5.symbol_info_tick(symbol) + if symbol_info_tick is None: + logger.error(f"无法获取{symbol}行情") + return + + price = symbol_info_tick.ask if order_type == "buy" else symbol_info_tick.bid + order_type_mt5 = mt5.ORDER_TYPE_BUY if order_type == "buy" else mt5.ORDER_TYPE_SELL + + request = { + "action": mt5.TRADE_ACTION_DEAL, + "symbol": symbol, + "volume": volume, + "type": order_type_mt5, + "price": price, + "deviation": 20, + "magic": 234000, + "comment": f"{order_type} order", + "type_filling": mt5.ORDER_FILLING_RETURN, + } + + result = mt5.order_send(request) + if result.retcode != mt5.TRADE_RETCODE_DONE: + logger.error(f"下单失败，retcode={result.retcode}") + else: + logger.info(f"下单成功: {order_type} {symbol} {volume}")