# -*- coding: utf-8 -*- import tushare as ts import time import pandas as pd import csv # Tushare token TUSHARE_TOKEN ee5c0e991e17949cdafbcf8ec42321ef4bac94e9ca3474e4d62313a3 ts.set_token(TUSHARE_TOKEN) pro ts.pro_api() # 获取申万三级行业列表 df_industry pro.index_classify(levelL3, srcSW2021) industry_list df_industry[[index_code, industry_name]].values.tolist() def get_latest_revenue_growth(ts_code: str): 获取个股最新营收同比、净利润同比 df pro.fina_indicator(ts_codets_code) time.sleep(0.35) if df.empty: return None, None, None # 按报告期倒序排列最新财报第一行 df df.sort_values(end_date, ascendingFalse).reset_index(dropTrue) latest df.iloc[0] report_date latest[end_date] tr_yoy latest[tr_yoy] # 营收同比 netprofit_yoy latest[netprofit_yoy] # 净利润同比 return report_date, tr_yoy, netprofit_yoy def get_stock_name(ts_code: str): 根据股票代码获取股票名称 df pro.stock_basic(ts_codets_code, fieldsts_code,name) time.sleep(0.2) if df.empty: return 未知股票 return df.iloc[0][name] def get_current_pe_ttm(ts_code: str): 获取PE-TTM空值返回None不会强制转float df pro.daily_basic( ts_codets_code, start_date20260722, fieldstrade_date,pe_ttm ) time.sleep(0.35) if df.empty: return None df df.sort_values(trade_date, ascendingFalse) latest df.iloc[0] pe latest[pe_ttm] # 是空/NaN 统一返回None if pd.isna(pe) or pe is None: return None return float(pe) # 全局变量表头只写一次 csv_path hangye.csv # 检测文件是否存在不存在则写入表头 try: with open(csv_path, r, encodingutf-8-sig) as f_check: pass except FileNotFoundError: # 文件不存在新建并写入表头 with open(csv_path, w, encodingutf-8-sig, newline) as f_head: writer_head csv.writer(f_head) writer_head.writerow([ ts_code, 股票名称, 营业收入同比增长率(%), 利润同比增长率(%), 市盈率PE-TTM, 所属行业 ]) index 0 # 遍历所有行业 for idx_code, idx_name in industry_list: index 1 print(f\n 正在处理行业{idx_name} 序号{index} ) # 获取行业内全部股票 stock_df pro.index_member_all(l3_codeidx_code) ts_codes stock_df[ts_code].unique().tolist() # 统计变量初始化每个行业独立清零 growth_count 0.0 profit_growth_count 0.0 count 0 # 营收有效样本数 count_profit 0 # 利润有效样本数 # 追加写入CSV with open(csv_path, a, encodingutf-8-sig, newline) as f: writer csv.writer(f) for ts_code in ts_codes: try: period, growth, profit_growth get_latest_revenue_growth(ts_code) name get_stock_name(ts_code) # 跳过未知股票 if 未知股票 in name: continue peTTM get_current_pe_ttm(ts_code) # 核心校验任意一项为空/NaN 直接跳过这支股票杜绝None参与运算 if (period is None or growth is None or pd.isna(growth) or profit_growth is None or pd.isna(profit_growth)): continue # 转为浮点保证数值类型统一 growth float(growth) profit_growth float(profit_growth) # 控制台打印信息 print(f{ts_code} {name:20} | 营收同比:{growth:6.2f}% | 净利同比:{profit_growth:6.2f}%) # 组装写入CSV的数据 pe_show round(peTTM, 2) if peTTM is not None else writer.writerow([ ts_code, name, round(growth, 2), round(profit_growth, 2), pe_show, idx_name ]) # 绝对值小于200才纳入平均值统计 if abs(growth) 200: growth_count growth count 1 if abs(profit_growth) 200: profit_growth_count profit_growth count_profit 1 except Exception as e: print(f{ts_code} 获取失败: {str(e)}) continue # 计算行业平均增长率 avg_growth growth_count / count if count 0 else 0.0 avg_profit_growth profit_growth_count / count_profit if count_profit 0 else 0.0 # 写入行业汇总行 writer.writerow([ 【行业均值汇总】, idx_name, round(avg_growth, 2), round(avg_profit_growth, 2), , f有效个股数量{count} ]) # 控制台打印行业汇总结果 print(f\n【{idx_name}汇总】) print(f行业平均营收同比{avg_growth:.2f} %) print(f行业平均净利同比{avg_profit_growth:.2f} %) print(f有效统计公司数{count}\n) time.sleep(1) # 每个行业结束停顿1秒防止接口请求过快被限制
