零基础 Python 量化选股工具|自动下载 A 股财报,一键剔除暴雷股(附完整源码)

# -*- coding: utf-8 -*-import syssys.path.append(r"F:\AI_Robot_Libs")import pickleimport pandas as pdimport numpy as npimport osimport timefrom datetime import datetimeimport tushare as tsprint("【日志】财务模块加载完成，初始化Tushare数据源")# ============ 仅需修改这里的TOKEN ============BASE_FOLDER = r"D:\tdx\finance_cache"CACHE_PATH = os.path.join(BASE_FOLDER, "fin_cache.pkl")QUERY_DELAY = 0.8MAX_RETRY = 3# 填入你自己的TOKENTUSHARE_TOKEN = "c49cd684f193bf9971b60894a51ac6511e5831499b39531705c56aaa"# 风控筛选参数，可自行调整THRESHOLD_ST = TrueTHRESHOLD_GOODWILL = 0.3THRESHOLD_PROFIT = -10# ============================================pro = ts.pro_api(TUSHARE_TOKEN)def sync_financial_data(retry_times = MAX_RETRY):    print(f"【日志】同步启动，剩余重试次数：{retry_times}")    if not os.path.exists(BASE_FOLDER):        os.makedirs(BASE_FOLDER)    print("===== 开始同步全市场财务数据 =====")    stock_all = pro.stock_basic(list_status='L', fields='ts_code,symbol,name,industry')    stock_all['is_st'] = np.where(stock_all['name'].str.contains('ST|*ST'), "是", "否")    code_list = stock_all['ts_code'].tolist()    all_fin_data = []    for idx, ts_code in enumerate(code_list):        try:            df = pro.fina_indicator(ts_code=ts_code, start_date='20250101')            if len(df) == 0:                continue            latest = df.sort_values("end_date", ascending=False).iloc[0]            info = stock_all[stock_all['ts_code']==ts_code].iloc[0]            gw = float(latest['goodwill']) if pd.notna(latest['goodwill']) else 0            eq = float(latest['total_hldr_eqy_exc_min_int']) if pd.notna(latest['total_hldr_eqy_exc_min_int']) else 1            row = {                "code":info['symbol'],                "name":info['name'],                "industry":info['industry'],                "roe":float(latest['roe']) if pd.notna(latest['roe']) else 0,                "pe_ttm":float(latest['pe_ttm']) if pd.notna(latest['pe_ttm']) else 999,                "net_profit_growth":float(latest['profit_yoy']) if pd.notna(latest['profit_yoy']) else -99,                "goodwill":gw,                "net_asset":eq,                "is_st":info['is_st'],                "goodwill_ratio": gw / eq            }            all_fin_data.append(row)        except Exception:            continue        time.sleep(QUERY_DELAY)        if idx % 200 == 0:            print(f"【进度】已处理 {idx}/{len(code_list)}")    df_save = pd.DataFrame(all_fin_data)    with open(CACHE_PATH,"wb") as f:        pickle.dump(df_save,f)    print("【日志】财务数据同步完成，已缓存")    return df_savedef check_cache_valid():    if not os.path.exists(CACHE_PATH):        return False    t = os.path.getmtime(CACHE_PATH)    diff = (datetime.now()-datetime.fromtimestamp(t)).total_seconds()/3600    return diff < 24def load_fin_cache():    with open(CACHE_PATH,"rb") as f:        return pickle.load(f)def filter_risk_stock(df):    mask = df["is_st"]!="是"    mask = mask & (df["goodwill_ratio"] < THRESHOLD_GOODWILL)    mask = mask & (df["net_profit_growth"] > THRESHOLD_PROFIT)    return df[mask].reset_index(drop=True)def calc_fin_score(row):    s1 = np.clip(row["roe"],-10,30)/40*40    s2 = np.clip(row["net_profit_growth"],-20,50)/70*30    s3 = np.clip(100/row["pe_ttm"],0,30) if row["pe_ttm"]>0 else 0    return round(s1+s2+s3,2)if __name__ == "__main__":    if not check_cache_valid():        sync_financial_data()    data = load_fin_cache()    safe = filter_risk_stock(data)    safe["基本面得分"] = safe.apply(calc_fin_score,axis=1)    print("\n========== 优质基本面股票清单（得分排序）==========")    print(safe[["code","name","industry","roe","pe_ttm","net_profit_growth","基本面得分"]].sort_values("基本面得分",ascending=False))