32岁零基础学Python量化 第5周 实战篇:用代码回测了网络热门“涨停回踩”策略,结果让我意外

import tushare as ts #国内主流财经数据接口import pandas as pdimport json  #Python 内置的标准库模块import time   #内置标准库用于处理时间相关任务下面用于控制请求频率避免触发Tushare限制from datetime import datetime, timedelta  #导入datetime,timedelta函数# ========== 配置Tushare Token(注册账号后免费获取)==========TS_TOKEN = "08f806d80734e73800cc55d0a9b28ca6bd58e42de9249feb10de6b0d"  # 登录 https://tushare.pro 获取ts.set_token(TS_TOKEN)pro = ts.pro_api()# ========== 辅助函数 ==========#1.获取股票日线数据def get_stock_daily(ts_code, years=2):    """    获取股票近 years 年的日线数据（包含 pre_close 和 pct_chg）    返回按交易日升序排列的 DataFrame    """    end_date = datetime.now().strftime("%Y%m%d")    start_date = (datetime.now() - timedelta(days=years * 365)).strftime("%Y%m%d")    # 获取日线数据    df = pro.daily(ts_code=ts_code,start_date=start_date,end_date=end_date)    if df.empty:        raise ValueError(f"未获取到 {ts_code} 的数据，请检查股票代码和网络。")    # 按交易日升序排列    df = df.sort_values("trade_date").reset_index(drop=True)    return df#2.筛选涨停日d0def is_limit_up(row, prev_close, limit_threshold=9.8):    """    简单涨停判断：    - 主板默认 9.8% 以上且收盘价等于最高价    - 科创板/创业板/北交所可根据代码调整 limit_threshold    """    if prev_close is None or prev_close == 0:        return False    pct = (row['close'] - prev_close) / prev_close * 100    # 同时使用 pct_chg 辅助判断    return (pct >= limit_threshold) and (row['close'] == row['high'])#3.找到涨停日后进行策略条件筛选def analyze_limit_up(ts_code, df, limit_threshold=9.8):    """    对涨停日进行条件回溯与预测统计    """    success_dates = [] #预测成功的日期    loss_dates = []    #失败的日期    success_time = 0  #总的预测成功次数    loss_time = 0     #总的预测失败次数    total_limit_up = 0  #统计总涨停次数    # 需要前一日收盘价来判断涨停，向前补充 pre_close（df 中已有 pre_close 字段）    # Tushare daily 数据包含 pre_close    n = len(df)    for i in range(n - 4):  # 必须保证有 d1~d4        row_d0 = df.iloc[i]        # 涨停判断        if not is_limit_up(row_d0, row_d0['pre_close'], limit_threshold):            continue        total_limit_up += 1  #统计总涨停次数        d0 = row_d0        d1 = df.iloc[i+1]        d2 = df.iloc[i+2]        d3 = df.iloc[i+3]        d4 = df.iloc[i+4]        #'涨停回踩'策略的四个条件        # 条件 a: d0>d1>d2>d3 收盘价三连阴        cond_a = (d0['close'] > d1['close'] > d2['close'] > d3['close'])        # 条件 b: 成交量 d1*0.9 > d2, d2*0.9 > d3        cond_b = (d1['vol'] * 0.9 > d2['vol']) and (d2['vol'] * 0.9 > d3['vol'])        # 条件 c: d1 最高价 > d0 收盘价        cond_c = d1['high'] > d0['close']        # 条件 d: d3 收盘价 > d0开盘价 + (d0收盘价-d0开盘价)/3        cond_d = d3['close'] > (d0['open'] + (d0['close'] - d0['open']) / 3)        #后续修改需排除一字板（实体为0）: if d0['close'] == d0['open']:continue        if cond_a and cond_b and cond_c and cond_d:            # 筛选满足涨停回踩策略的涨停日并更新成功次数和失败次数以及对应日期            d4_cond = (d4['high'] - d3['close']) / d3['close'] >= 0.05            if d4_cond:                success_time += 1                success_dates.append(d0['trade_date'])            else:                loss_time += 1                loss_dates.append(d0['trade_date'])    # 计算成功率    total = success_time + loss_time    success_rate = success_time / total if total > 0 else 0.0    # 整理结果    result = {        'stock_code': ts_code,        'success_time': success_time,        'success_dates': ','.join(success_dates),        'loss_time': loss_time,        'loss_dates': ','.join(loss_dates),        'success_rate': round(success_rate, 4),        'total_limit_up':total_limit_up    }    return result#4.获取大A主板非ST股票代码列表def get_mainboard_stocks():    """    使用读取本地txt股票代码文件的方式，需准备好所有主板股票代码的txt文件，有需要可以私信我获取    """    ts_code_list = []   #新建股票代码列表    file_name = "stock_codes.txt"  #股票代码的txt文件的文件名    try:        # 读取本地文本，按行解析代码        with open(file_name, "r", encoding="utf-8") as f:            lines = f.readlines()            for line in lines:                code = line.strip()                if not code:                    continue                # 二次校验主板规则 + 拼接Tushare标准ts_code(需包含后缀上交：SH,深交：SZ)                if code.startswith("60"):                    # 上交所主板 .SH                    ts_code_list.append(f"{code}.SH")                elif code.startswith(("000", "001", "002")):                    # 深交所主板 .SZ                    ts_code_list.append(f"{code}.SZ")        # 去重、排序        ts_code_list = list(set(ts_code_list))        ts_code_list.sort()        print(f"从本地文件读取到主板股票总数：{len(ts_code_list)}")        print("前10个Tushare代码：", ts_code_list[:10])        return ts_code_list    #预防找不到文件以及读取异常    except FileNotFoundError:        print(f"错误：未找到 {file_name}，请确认文件和代码在同一目录！")        return []    except Exception as e:        print(f"读取文件异常：{e}")        return []#5.使用给定策略分析单只股票def analyze_stock(ts_code, limit_threshold=9.8):    """    对单只股票执行完整分析流程，并返回结果字典；    若出现异常，打印错误信息并返回 None。    """    try:        print(f"正在分析 {ts_code} ...")        df = get_stock_daily(ts_code, years=2)        summary = analyze_limit_up(ts_code, df, limit_threshold)        return summary    except Exception as e:        print(f"股票 {ts_code} 分析失败，错误信息：{e}")        return None#6.使用for循环循环分析txt文件内的所有股票代码对应日线的数据def batch_analyze_all(limit_threshold=9.8):    """    批量分析所有主板非ST股票，汇总结果并计算总成功率    """    #1.获取A股主板非ST股票代码赋值给列表stock_list    stock_list = get_mainboard_stocks()    results = []    #2.对stock_list进行循环，获取单个股票代码    for idx, stock in enumerate(stock_list, 1):        print(f"进度：{idx}/{len(stock_list)} --- 股票：{stock}")        #3.进行涨停分析        res = analyze_stock(stock, limit_threshold)        if res is not None:            results.append(res)        # 控制请求频率，避免触发 Tushare 限制        time.sleep(1.1)    # 4.汇总所有结果    if not results:        print("没有成功分析的股票数据，无法生成汇总报告。")        return    df_all = pd.DataFrame(results)    # 5.计算总成功率    total_success = df_all['success_time'].sum()    total_loss = df_all['loss_time'].sum()    total_rate = total_success / (total_success + total_loss) if (total_success + total_loss) > 0 else 0.0    print(f"\n========== 全市场汇总 ==========")    print(f"总成功次数：{total_success}")    print(f"总失败次数：{total_loss}")    print(f"总成功率：{total_rate:.4f}")    # 6.保存详细结果    detail_file = "all_stocks_analysis.csv"    df_all.to_csv(detail_file, index=False, encoding='utf-8-sig')    print(f"各股票详细分析结果已保存至：{detail_file}")    # 7.保存总成功率    overall_df = pd.DataFrame([{        'total_success': total_success,        'total_loss': total_loss,        'overall_success_rate': round(total_rate, 4)    }])    overall_file = "overall_summary.csv"    overall_df.to_csv(overall_file, index=False, encoding='utf-8-sig')    print(f"总成功率汇总已保存至：{overall_file}")#主函数if __name__ == "__main__":    # 涨停阈值：主板 9.8    batch_analyze_all(limit_threshold=9.8)