Python金融实战:精讲红买绿停策略交易技巧

import osimport pandas as pdimport numpy as npimport structimport matplotlib.pyplot as pltfrom tqdm import tqdm# 解析通达信.day文件（核心函数）def parse_tdx_day(file_path):    """    解析通达信二进制.day文件，返回K线DataFrame    参数：file_path: .day文件路径    返回：包含日期、开盘价、最高价、最低价、收盘价、成交量、成交额的DataFrame    """    try:        with open(file_path, 'rb') as f:            buffer = f.read()        # 通达信.day格式：每32字节一条数据，包含8个字段        data_num = len(buffer) // 32  # 计算数据条数        k_list = []        for i in range(data_num):            off = i * 32            # 解包二进制数据，对应字段：日期、开盘价、最高价、最低价、收盘价、成交额、成交量、保留字段            date, open_, high, low, close, amount, volume, _ = struct.unpack('IIIIIfII', buffer[off:off+32])            # 处理数据格式：日期转为datetime，价格/成交量除以对应系数（通达信存储格式）            k_list.append([                pd.to_datetime(str(date)),  # 日期：YYYYMMDD转为datetime                open_/100,  # 开盘价：通达信存储为整数，需除以100                high/100,   # 最高价：除以100                low/100,    # 最低价：除以100                close/100,  # 收盘价：除以100                volume,     # 成交量                amount      # 成交额            ])        # 构建DataFrame并设置索引        df = pd.DataFrame(k_list, columns=['date', 'open', 'high', 'low', 'close', 'volume', 'amount'])        df.set_index('date', inplace=True)        df.dropna(inplace=True)        return df if len(df) > 60 else None  # 过滤数据不足60天的股票（确保均线计算有效）    except Exception as e:        print(f"解析文件{file_path}失败：{e}")        return None

class RedGreenStrategy:    def __init__(self, X1=35, X2=35, X3=3):        # 策略参数（对应通达信公式中的X_1~X_3，可自定义调整）        self.X1 = X1  # HHV/LLV的周期        self.X2 = X2  # X_7的偏移值        self.X3 = X3  # X_13的阈值    def calculate_signals(self, df):        """        计算红买、绿停、黄观望信号        参数：df: 股票K线DataFrame（包含open、high、low、close等字段）        返回：添加了信号的DataFrame        """        data = df.copy()        close = data['close']        high = data['high']        low = data['low']        # 1. 计算基础均线（对应公式X_4~X_6、X_16~X_17）        data['ma5'] = close.rolling(5).mean()    # X_4: 5日均线        data['ma10'] = close.rolling(10).mean()  # X_5: 10日均线        data['ma20'] = close.rolling(20).mean()  # X_6: 20日均线        data['ma13'] = close.rolling(13).mean()  # 13日均线        data['ma34'] = close.rolling(34).mean()  # 34日均线        data['X16'] = data[['ma13', 'ma34']].max(axis=1)  # X_16: 13/34均线最大值        data['X17'] = data[['ma13', 'ma34']].min(axis=1)  # X_17: 13/34均线最小值        # 2. 计算震荡辅助指标（对应公式X_7~X_13，用于过滤信号）        hhv_x1 = high.rolling(self.X1).max()     # X_1周期内的最高价        llv_x1 = low.rolling(self.X1).min()      # X_1周期内的最低价        data['X7'] = (hhv_x1 - close) / (hhv_x1 - llv_x1) * 100 - self.X2        data['X8'] = data['X7'].rolling(self.X1).mean() + 100  # SMA平滑（用rolling mean近似）        data['X9'] = (close - llv_x1) / (hhv_x1 - llv_x1) * 100        data['X10'] = data['X9'].rolling(3).mean()        data['X11'] = data['X10'].rolling(3).mean() + 100        data['X12'] = data['X11'] - data['X8']        data['X13'] = np.where(data['X12'] > self.X3, data['X12'] - self.X3, 0) * 2.5        # 3. 生成核心交易信号（红买、绿停、黄观望）        data['红买信号'] = close < data['X17']    # 红买：收盘价<X17        data['绿停信号'] = close > data['X16']    # 绿停：收盘价>X16        data['黄观望信号'] = (data['X16'] >= close) & (close >= data['X17'])  # 黄观望：介于两者之间        # 4. 生成标准化交易信号（1=买入，-1=卖出，0=持有/观望），便于回测        data['交易信号'] = 0        data.loc[data['红买信号'], '交易信号'] = 1    # 买入信号        data.loc[data['绿停信号'], '交易信号'] = -1   # 卖出信号        # 去除空值，确保回测有效性        data.dropna(inplace=True)        return data

def plot_signals(df, stock_code):    """    可视化股票K线、均线和交易信号    参数：df: 带信号的DataFrame；stock_code: 股票代码    """    plt.rcParams['font.sans-serif'] = ['SimHei']  # 解决中文显示问题    plt.rcParams['axes.unicode_minus'] = False    # 解决负号显示问题    fig, ax = plt.subplots(figsize=(12, 6))    # 绘制K线（简化版，用收盘价线代替）    ax.plot(df.index, df['close'], color='gray', linewidth=1.5, label='收盘价')    # 绘制13/34日均线（策略核心均线）    ax.plot(df.index, df['ma13'], color='orange', linewidth=1.2, label='13日均线')    ax.plot(df.index, df['ma34'], color='blue', linewidth=1.2, label='34日均线')    # 标注红买、绿停信号    buy_signals = df[df['红买信号']]    sell_signals = df[df['绿停信号']]    ax.scatter(buy_signals.index, buy_signals['close'], marker='^', color='red', s=100, label='红买信号')    ax.scatter(sell_signals.index, sell_signals['close'], marker='v', color='green', s=100, label='绿停信号')    # 图表设置    ax.set_title(f'{stock_code} 红买绿停策略信号可视化', fontsize=14)    ax.set_ylabel('价格（元）', fontsize=12)    ax.legend(loc='best')    ax.grid(True, alpha=0.3)    plt.xticks(rotation=45)    plt.tight_layout()    plt.show()# 测试：单只股票信号生成与可视化if __name__ == "__main__":    # 1. 读取单只股票的.day文件（替换为你的通达信文件路径）    stock_file = r"D:\软件安装\通达信\vipdoc\sz\lday\sz000001.day"  # 平安银行    k_data = parse_tdx_day(stock_file)    if k_data is None:        print("数据解析失败，请检查文件路径或文件完整性")        exit()    # 2. 生成策略信号    strategy = RedGreenStrategy()    signal_df = strategy.calculate_signals(k_data)    # 3. 可视化信号    plot_signals(signal_df, "sz000001（平安银行）")

class StrategyBacktester:    def __init__(self, initial_capital=1000000, fee_rate=0.0003):        """        回测引擎初始化        参数：initial_capital: 初始资金（默认100万）；fee_rate: 手续费（默认万3，贴合A股实盘）        """        self.initial_capital = initial_capital        self.fee_rate = fee_rate    def backtest_single_stock(self, df, stock_code):        """        单只股票回测，计算核心评估因子        参数：df: 带信号的DataFrame；stock_code: 股票代码        返回：回测结果字典（包含收益率、最大回撤等核心指标）        """        try:            data = df.copy()            # 生成持仓信号：信号次日执行（避免未来函数）            data['持仓'] = data['交易信号'].replace(0, np.nan).ffill().fillna(0)            data['持仓'] = data['持仓'].shift(1).fillna(0)            # 计算每日收益            data['每日收益率'] = data['close'].pct_change().fillna(0)            data['策略收益率'] = data['持仓'] * data['每日收益率']            # 扣除手续费（每次交易收取，买卖都扣）            trade_changes = data['持仓'].diff().fillna(0).abs()  # 持仓变化（买卖动作）            data['策略收益率'] -= trade_changes * self.fee_rate            # 计算累计资产和核心评估因子            data['累计资产'] = self.initial_capital * (1 + data['策略收益率']).cumprod()            final_capital = data['累计资产'].iloc[-1]            # 核心评估因子（量化实战必看）            total_return = (final_capital / self.initial_capital) - 1  # 总收益率            annual_return = total_return * 252 / len(data)             # 年化收益率（252个交易日）            # 最大回撤（风险核心指标）            data['累计最大值'] = data['累计资产'].cummax()            data['回撤'] = (data['累计资产'] - data['累计最大值']) / data['累计最大值']            max_drawdown = data['回撤'].min()            # 交易次数、胜率、盈亏比            trades = data[data['交易信号'] != 0]            trade_returns = []            buy_price = None            for _, row in data.iterrows():                if row['交易信号'] == 1:                    buy_price = row['close']                elif row['交易信号'] == -1 and buy_price is not None:                    trade_return = (row['close'] - buy_price) / buy_price                    trade_returns.append(trade_return)                    buy_price = None            trade_count = len(trade_returns)            win_rate = len([x for x in trade_returns if x > 0]) / trade_count if trade_count > 0 else 0            # 盈亏比：平均盈利/平均亏损            profit_returns = [x for x in trade_returns if x > 0]            loss_returns = [abs(x) for x in trade_returns if x < 0]            avg_profit = np.mean(profit_returns) if profit_returns else 0            avg_loss = np.mean(loss_returns) if loss_returns else 0            profit_loss_ratio = avg_profit / avg_loss if avg_loss != 0 else 0            # 夏普比率（风险调整后收益，>1合格，>2优秀）            sharpe_ratio = np.mean(data['策略收益率']) / np.std(data['策略收益率']) * np.sqrt(252) if np.std(data['策略收益率']) != 0 else 0            return {                '股票代码': stock_code,                '初始资金': self.initial_capital,                '最终资金': round(final_capital, 2),                '总收益率(%)': round(total_return * 100, 2),                '年化收益率(%)': round(annual_return * 100, 2),                '最大回撤(%)': round(max_drawdown * 100, 2),                '交易次数': trade_count,                '胜率(%)': round(win_rate * 100, 2),                '盈亏比': round(profit_loss_ratio, 2),                '夏普比率': round(sharpe_ratio, 2),                '数据天数': len(data)            }        except Exception as e:            print(f"回测{stock_code}失败：{e}")            return None# 批量回测主函数def batch_backtest(data_paths, output_dir='./红买绿停回测结果'):    """    批量回测指定路径下的所有.day文件    参数：data_paths: 通达信.day文件所在目录（可传入多个）；output_dir: 回测结果保存目录    """    # 创建结果保存目录    os.makedirs(output_dir, exist_ok=True)    # 初始化工具    backtester = StrategyBacktester()    strategy = RedGreenStrategy()    result_list = []    # 扫描所有.day文件    for data_path in data_paths:        if not os.path.exists(data_path):            print(f"路径不存在：{data_path}")            continue        print(f"正在扫描目录：{data_path}")        day_files = [f for f in os.listdir(data_path) if f.endswith('.day')]        # 批量处理每个文件        for filename in tqdm(day_files, desc=f"处理{os.path.basename(data_path)}"):            file_path = os.path.join(data_path, filename)            stock_code = filename.replace('.day', '')  # 提取股票代码（如sh600000）            # 解析数据、生成信号、执行回测            k_data = parse_tdx_day(file_path)            if k_data is None:                continue            signal_df = strategy.calculate_signals(k_data)            if len(signal_df) < 30:                continue            backtest_result = backtester.backtest_single_stock(signal_df, stock_code)            if backtest_result:                result_list.append(backtest_result)    # 保存回测汇总报告    result_df = pd.DataFrame(result_list)    result_csv = os.path.join(output_dir, '红买绿停策略回测汇总.csv')    result_df.to_csv(result_csv, index=False, encoding='utf-8-sig')    print(f"\n批量回测完成！汇总报告已保存至：{result_csv}")    # 打印策略整体评估    if not result_df.empty:        print("\n==================== 策略整体评估 ====================")        print(f"回测股票总数：{len(result_df)} 只")        print(f"平均总收益率：{round(result_df['总收益率(%)'].mean(), 2)}%")        print(f"平均年化收益率：{round(result_df['年化收益率(%)'].mean(), 2)}%")        print(f"平均最大回撤：{round(result_df['最大回撤(%)'].mean(), 2)}%")        print(f"平均胜率：{round(result_df['胜率(%)'].mean(), 2)}%")        print(f"平均盈亏比：{round(result_df['盈亏比'].mean(), 2)}")        print(f"平均夏普比率：{round(result_df['夏普比率'].mean(), 2)}")    else:        print("未完成任何有效回测，请检查数据路径或文件完整性")# 执行批量回测（替换为你的通达信数据路径）if __name__ == "__main__":    sz_data_path = r"D:\软件安装\通达信\vipdoc\sz\lday"  # 深圳股票数据路径    sh_data_path = r"D:\软件安装\通达信\vipdoc\sh\lday"  # 上海股票数据路径    batch_backtest([sz_data_path, sh_data_path])

# 在calculate_signals函数中添加成交量过滤# 新增：成交量大于近5日平均成交量的1.2倍，才确认信号有效data['5日平均成交量'] = data['volume'].rolling(5).mean()data['红买信号'] = (close < data['X17']) & (data['volume'] > data['5日平均成交量'] * 1.2)data['绿停信号'] = (close > data['X16']) & (data['volume'] > data['5日平均成交量'] * 1.2)