量化择时实战:Python识别买卖点,让你的策略更精准!

import tushare as tsimport pandas as pdimport matplotlib.pyplot as pltimport numpy as npfrom datetime import datetime, timedelta# 设置中文字体plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = False# 初始化tusharets.set_token('你的token')pro = ts.pro_api()# 获取多只股票数据，便于对比分析def get_stock_data(ts_code, start_date, end_date):    df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)    df['trade_date'] = pd.to_datetime(df['trade_date'])    df = df.sort_values('trade_date')    return df# 获取沪深300成分股中不同行业的代表股票stocks = {    '贵州茅台': '600519.SH',    '中国平安': '601318.SH',     '宁德时代': '300750.SZ',    '招商银行': '600036.SH'}start_date = '20230101'end_date = '20240601'stock_data = {}for name, code in stocks.items():    stock_data[name] = get_stock_data(code, start_date, end_date)    print(f"{name} 数据量: {len(stock_data[name])}")

def calculate_technical_indicators(df):    """计算所有技术指标"""    # 移动平均线    df['MA5'] = df['close'].rolling(5).mean()    df['MA10'] = df['close'].rolling(10).mean()     df['MA20'] = df['close'].rolling(20).mean()    df['MA60'] = df['close'].rolling(60).mean()    # MACD指标    df['EMA12'] = df['close'].ewm(span=12, adjust=False).mean()    df['EMA26'] = df['close'].ewm(span=26, adjust=False).mean()    df['DIF'] = df['EMA12'] - df['EMA26']    df['DEA'] = df['DIF'].ewm(span=9, adjust=False).mean()    df['MACD'] = (df['DIF'] - df['DEA']) * 2    # RSI指标    delta = df['close'].diff()    gain = delta.where(delta > 0, 0)    loss = -delta.where(delta < 0, 0)    avg_gain = gain.rolling(14).mean()    avg_loss = loss.rolling(14).mean()    rs = avg_gain / avg_loss    df['RSI'] = 100 - (100 / (1 + rs))    # 成交量均线    df['VOL_MA5'] = df['vol'].rolling(5).mean()    return df# 为所有股票计算技术指标for name in stocks.keys():    stock_data[name] = calculate_technical_indicators(stock_data[name])

def identify_trading_signals(df):    """识别买卖信号"""    df = df.copy()    df['Signal'] = 0  # 0:持有, 1:买入, -1:卖出    # 买入条件（同时满足）：    # 1. 价格上穿20日均线    # 2. MACD金叉（DIF上穿DEA）    # 3. RSI脱离超卖区（>30）    buy_condition = (        (df['close'] > df['MA20']) &         (df['close'].shift(1) <= df['MA20'].shift(1)) &        (df['DIF'] > df['DEA']) &        (df['DIF'].shift(1) <= df['DEA'].shift(1)) &        (df['RSI'] > 30)    )    # 卖出条件（满足任一）：    # 1. 价格下穿20日均线    # 2. MACD死叉    # 3. RSI超买（>70）    sell_condition = (        (df['close'] < df['MA20']) &         (df['close'].shift(1) >= df['MA20'].shift(1))    ) | (        (df['DIF'] < df['DEA']) &         (df['DIF'].shift(1) >= df['DEA'].shift(1))    ) | (        df['RSI'] > 70    )    df.loc[buy_condition, 'Signal'] = 1    df.loc[sell_condition, 'Signal'] = -1    return df# 应用信号识别for name in stocks.keys():    stock_data[name] = identify_trading_signals(stock_data[name])

def backtest_strategy(df, initial_capital=100000):    """策略回测"""    df = df.copy()    # 初始化仓位和资金列    df['Position'] = 0  # 持仓数量    df['Cash'] = initial_capital  # 现金    df['Total'] = initial_capital  # 总资产    df['Returns'] = 0.0  # 每日收益率    position = 0  # 当前持仓    cash = initial_capital  # 当前现金    for i in range(1, len(df)):        current_signal = df['Signal'].iloc[i]        current_price = df['close'].iloc[i]        # 买入信号且空仓        if current_signal == 1 and position == 0:            # 全仓买入            position = cash // current_price            cash -= position * current_price * 1.0003  # 考虑佣金        # 卖出信号且持仓        elif current_signal == -1 and position > 0:            # 全仓卖出            cash += position * current_price * 0.9997  # 考虑佣金            position = 0        # 更新账户信息        df.iloc[i, df.columns.get_loc('Position')] = position        df.iloc[i, df.columns.get_loc('Cash')] = cash        df.iloc[i, df.columns.get_loc('Total')] = cash + position * current_price        # 计算收益率        if i > 1:            prev_total = df['Total'].iloc[i-1]            current_total = df['Total'].iloc[i]            df.iloc[i, df.columns.get_loc('Returns')] = (current_total - prev_total) / prev_total    return df# 执行回测backtest_results = {}for name in stocks.keys():    backtest_results[name] = backtest_strategy(stock_data[name])

def plot_strategy_performance(df, stock_name):    """绘制策略绩效图"""    fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(15, 12))    # 股价和买卖点    ax1.plot(df['trade_date'], df['close'], label='收盘价', linewidth=2)    ax1.plot(df['trade_date'], df['MA20'], label='20日均线', alpha=0.7)    # 标记买卖点    buy_signals = df[df['Signal'] == 1]    sell_signals = df[df['Signal'] == -1]    ax1.scatter(buy_signals['trade_date'], buy_signals['close'],                color='red', marker='^', s=100, label='买入信号', zorder=5)    ax1.scatter(sell_signals['trade_date'], sell_signals['close'],               color='green', marker='v', s=100, label='卖出信号', zorder=5)    ax1.set_title(f'{stock_name} - 股价走势与买卖点', fontsize=14)    ax1.legend()    ax1.grid(True, alpha=0.3)    # 技术指标    ax2.plot(df['trade_date'], df['DIF'], label='DIF', linewidth=1.5)    ax2.plot(df['trade_date'], df['DEA'], label='DEA', linewidth=1.5)    ax2.bar(df['trade_date'], df['MACD'], label='MACD', alpha=0.5)    ax2.set_title('MACD指标', fontsize=14)    ax2.legend()    ax2.grid(True, alpha=0.3)    # 资产曲线    ax3.plot(df['trade_date'], df['Total'], label='策略总资产', linewidth=2)    ax3.plot(df['trade_date'], df['close'] / df['close'].iloc[0] * 100000,             label='买入持有', linewidth=2, alpha=0.7)    ax3.set_title('策略收益对比', fontsize=14)    ax3.legend()    ax3.grid(True, alpha=0.3)    plt.tight_layout()    plt.show()# 绘制各股票策略表现for name in stocks.keys():    plot_strategy_performance(backtest_results[name], name)

def calculate_performance_metrics(df, stock_name):    """计算策略绩效指标"""    returns = df['Returns'].dropna()    total_return = (df['Total'].iloc[-1] - 100000) / 100000    buy_hold_return = (df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0]    # 年化收益率    days = (df['trade_date'].iloc[-1] - df['trade_date'].iloc[0]).days    annual_return = (1 + total_return) ** (365 / days) - 1    # 最大回撤    cumulative = (1 + returns).cumprod()    peak = cumulative.expanding().max()    drawdown = (cumulative - peak) / peak    max_drawdown = drawdown.min()    # 夏普比率（假设无风险利率3%）    excess_returns = returns - 0.03/252    sharpe_ratio = excess_returns.mean() / excess_returns.std() * np.sqrt(252)    # 交易次数    trade_count = len(df[df['Signal'] != 0])    print(f"\n{stock_name}策略绩效:")    print(f"总收益率: {total_return:.2%}")    print(f"买入持有收益率: {buy_hold_return:.2%}")    print(f"年化收益率: {annual_return:.2%}")    print(f"最大回撤: {max_drawdown:.2%}")    print(f"夏普比率: {sharpe_ratio:.2f}")    print(f"交易次数: {trade_count}次")    return {        '总收益率': total_return,        '年化收益率': annual_return,        '最大回撤': max_drawdown,        '夏普比率': sharpe_ratio,        '交易次数': trade_count    }# 计算所有股票绩效performance_summary = {}for name in stocks.keys():    performance_summary[name] = calculate_performance_metrics(backtest_results[name], name)

def optimize_parameters(df):    """简单的参数优化示例"""    best_sharpe = -999    best_params = {}    # 测试不同的均线周期组合    for ma_short in [5, 10, 20]:        for ma_long in [20, 30, 60]:            # 这里可以添加参数优化的具体逻辑            # 由于篇幅限制，暂不展开详细优化过程            pass    return best_params

def add_risk_management(df, stop_loss=0.08):    """添加止损机制"""    df = df.copy()    df['Stop_Loss_Price'] = 0.0    entry_price = 0    for i in range(1, len(df)):        if df['Signal'].iloc[i] == 1 and df['Position'].iloc[i-1] == 0:            # 记录买入价格            entry_price = df['close'].iloc[i]            df.iloc[i, df.columns.get_loc('Stop_Loss_Price')] = entry_price * (1 - stop_loss)        elif df['Position'].iloc[i] > 0 and df['close'].iloc[i] < df['Stop_Loss_Price'].iloc[i]:            # 触发止损            df.iloc[i, df.columns.get_loc('Signal')] = -1    return df