From 559d78388d30aa62c783cfdb7cfec0d9c95629d9 Mon Sep 17 00:00:00 2001 From: songkunling Date: Fri, 25 Jul 2025 17:35:01 +0800 Subject: [PATCH] add files --- .gitignore | 4 ++ backtest.py | 45 +++++++++++++ config.py | 24 +++++++ logger.py | 40 +++++++++++ main.py | 90 +++++++++++++++++++++++++ optimizer.py | 115 ++++++++++++++++++++++++++++++++ requirements.txt | 3 + strategies/bollinger.py | 57 ++++++++++++++++ strategies/daily_breakout.py | 68 +++++++++++++++++++ strategies/kdj.py | 60 +++++++++++++++++ strategies/ma_cross.py | 56 ++++++++++++++++ strategies/macd.py | 58 ++++++++++++++++ strategies/mean_reversion.py | 57 ++++++++++++++++ strategies/momentum_breakout.py | 54 +++++++++++++++ strategies/profit_protect.py | 79 ++++++++++++++++++++++ strategies/resilient_trend.py | 113 +++++++++++++++++++++++++++++++ strategies/rsi.py | 59 ++++++++++++++++ strategies/turtle.py | 54 +++++++++++++++ utils.py | 70 +++++++++++++++++++ 19 files changed, 1106 insertions(+) create mode 100644 .gitignore create mode 100644 backtest.py create mode 100644 config.py create mode 100644 logger.py create mode 100644 main.py create mode 100644 optimizer.py create mode 100644 requirements.txt create mode 100644 strategies/bollinger.py create mode 100644 strategies/daily_breakout.py create mode 100644 strategies/kdj.py create mode 100644 strategies/ma_cross.py create mode 100644 strategies/macd.py create mode 100644 strategies/mean_reversion.py create mode 100644 strategies/momentum_breakout.py create mode 100644 strategies/profit_protect.py create mode 100644 strategies/resilient_trend.py create mode 100644 strategies/rsi.py create mode 100644 strategies/turtle.py create mode 100644 utils.py 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}")