793 lines
37 KiB
Python
Executable File
793 lines
37 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
TuShare 5年历史数据全面获取器
|
|
获取所有股票的基本面、技术指标、公司信息、股东结构、基金、融资、公告、分红配股等数据
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import logging
|
|
import pandas as pd
|
|
import tudata as ts
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Tuple
|
|
import numpy as np
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import sqlite3
|
|
import pymysql
|
|
from config_manager import get_config
|
|
|
|
class ComprehensiveDataFetcher:
|
|
"""全面数据获取器"""
|
|
|
|
def __init__(self, token: str = None):
|
|
self.config = get_config()
|
|
self.token = token or self.config.tushare.token
|
|
|
|
if not self.token:
|
|
raise ValueError("TuShare token is required")
|
|
|
|
ts.set_token(self.token)
|
|
self.pro = ts.pro_api()
|
|
|
|
# 设置日志
|
|
self.setup_logging()
|
|
|
|
# 数据库连接参数
|
|
self.db_config = {
|
|
'host': self.config.database.host,
|
|
'port': self.config.database.port,
|
|
'user': self.config.database.user,
|
|
'password': self.config.database.password,
|
|
'database': self.config.database.database,
|
|
'charset': self.config.database.charset
|
|
}
|
|
|
|
# 时间范围 (过去5年)
|
|
self.end_date = datetime.now().strftime('%Y%m%d')
|
|
self.start_date = (datetime.now() - timedelta(days=365*5)).strftime('%Y%m%d')
|
|
|
|
# 进度记录
|
|
self.progress_file = 'logs/comprehensive_5year_progress.json'
|
|
self.progress = self.load_progress()
|
|
|
|
# API限制配置
|
|
self.rate_limit = self.config.tushare.rate_limit # 每分钟请求数
|
|
self.batch_size = 50 # 批次大小
|
|
self.max_workers = 10 # 最大线程数
|
|
|
|
self.logger.info(f"🚀 初始化数据获取器 - 时间范围: {self.start_date} ~ {self.end_date}")
|
|
|
|
def setup_logging(self):
|
|
"""设置日志"""
|
|
os.makedirs('logs', exist_ok=True)
|
|
|
|
log_file = f'logs/comprehensive_5year_fetch_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler(log_file, encoding='utf-8'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def load_progress(self) -> Dict:
|
|
"""加载进度记录"""
|
|
if os.path.exists(self.progress_file):
|
|
try:
|
|
with open(self.progress_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except:
|
|
pass
|
|
|
|
return {
|
|
'stocks': [],
|
|
'completed_categories': [],
|
|
'last_update': None,
|
|
'total_stocks': 0,
|
|
'processed_stocks': 0
|
|
}
|
|
|
|
def save_progress(self):
|
|
"""保存进度记录"""
|
|
self.progress['last_update'] = datetime.now().isoformat()
|
|
os.makedirs(os.path.dirname(self.progress_file), exist_ok=True)
|
|
|
|
with open(self.progress_file, 'w', encoding='utf-8') as f:
|
|
json.dump(self.progress, f, ensure_ascii=False, indent=2)
|
|
|
|
def get_stock_list(self) -> pd.DataFrame:
|
|
"""获取股票列表"""
|
|
self.logger.info("📋 获取股票列表...")
|
|
|
|
try:
|
|
# 获取所有股票基本信息
|
|
stocks = self.pro.stock_basic(
|
|
fields='ts_code,symbol,name,area,industry,market,list_date,is_hs'
|
|
)
|
|
|
|
# 过滤有效股票 (上市时间超过1年)
|
|
cutoff_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d')
|
|
stocks = stocks[stocks['list_date'] <= cutoff_date]
|
|
|
|
# 排除ST和退市股票
|
|
stocks = stocks[
|
|
(~stocks['name'].str.contains('ST', na=False)) &
|
|
(~stocks['name'].str.contains('退', na=False)) &
|
|
(~stocks['name'].str.contains(r'\*', na=False, regex=True))
|
|
]
|
|
|
|
self.logger.info(f"✅ 获取股票列表成功: {len(stocks)}只股票")
|
|
return stocks
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"❌ 获取股票列表失败: {e}")
|
|
return pd.DataFrame()
|
|
|
|
def fetch_daily_data(self, ts_code: str) -> Optional[pd.DataFrame]:
|
|
"""获取日线数据"""
|
|
try:
|
|
time.sleep(0.1) # API限流
|
|
|
|
data = self.pro.daily(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date,
|
|
end_date=self.end_date,
|
|
fields='ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount'
|
|
)
|
|
|
|
if not data.empty:
|
|
data['fetch_time'] = datetime.now()
|
|
self.logger.debug(f" 📊 日线数据: {len(data)}条记录")
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取日线数据失败: {e}")
|
|
return None
|
|
|
|
def fetch_weekly_monthly_data(self, ts_code: str) -> Dict[str, pd.DataFrame]:
|
|
"""获取周线和月线数据"""
|
|
result = {}
|
|
|
|
try:
|
|
# 周线数据
|
|
time.sleep(0.1)
|
|
weekly = self.pro.weekly(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date,
|
|
end_date=self.end_date,
|
|
fields='ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount'
|
|
)
|
|
if not weekly.empty:
|
|
weekly['fetch_time'] = datetime.now()
|
|
result['weekly'] = weekly
|
|
self.logger.debug(f" 📊 周线数据: {len(weekly)}条记录")
|
|
|
|
# 月线数据
|
|
time.sleep(0.1)
|
|
monthly = self.pro.monthly(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date,
|
|
end_date=self.end_date,
|
|
fields='ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount'
|
|
)
|
|
if not monthly.empty:
|
|
monthly['fetch_time'] = datetime.now()
|
|
result['monthly'] = monthly
|
|
self.logger.debug(f" 📊 月线数据: {len(monthly)}条记录")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取周/月线数据失败: {e}")
|
|
|
|
return result
|
|
|
|
def fetch_financial_data(self, ts_code: str) -> Dict[str, pd.DataFrame]:
|
|
"""获取财务数据"""
|
|
result = {}
|
|
|
|
try:
|
|
# 利润表
|
|
time.sleep(0.15)
|
|
income = self.pro.income(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231',
|
|
fields='ts_code,ann_date,f_ann_date,end_date,report_type,comp_type,basic_eps,diluted_eps,total_revenue,revenue,int_income,prem_earned,comm_income,n_commis_income,n_oth_income,n_oth_b_income,prem_income,out_prem,une_prem_reser,reins_income,n_sec_tb_income,n_sec_uw_income,n_asset_mg_income,oth_b_income,fv_value_chg_gain,invest_income,ass_invest_income,forex_gain,total_cogs,oper_cost,int_exp,comm_exp,biz_tax_surchg,sell_exp,admin_exp,fin_exp,assets_impair_loss,prem_refund,compens_payout,reser_insur_liab,div_payt,reins_exp,oper_exp,compens_payout_refu,insur_reser_refu,reins_cost_refund,other_bus_cost,operate_profit,non_oper_income,non_oper_exp,nca_disploss,total_profit,income_tax,n_income,n_income_attr_p,minority_gain,oth_compr_income,t_compr_income,compr_inc_attr_p,compr_inc_attr_m_s,ebit,ebitda,insurance_exp,undist_profit,distable_profit'
|
|
)
|
|
if not income.empty:
|
|
income['fetch_time'] = datetime.now()
|
|
result['income'] = income
|
|
self.logger.debug(f" 💰 利润表: {len(income)}条记录")
|
|
|
|
# 资产负债表
|
|
time.sleep(0.15)
|
|
balancesheet = self.pro.balancesheet(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231',
|
|
fields='ts_code,ann_date,f_ann_date,end_date,report_type,comp_type,total_share,cap_rese,undistr_porfit,surplus_rese,special_rese,money_cap,trad_asset,notes_receiv,accounts_receiv,oth_receiv,prepayment,div_receiv,int_receiv,inventories,amor_exp,nca_within_1y,sett_rsrv,loanto_oth_bank_fi,premium_receiv,reinsur_receiv,reinsur_res_receiv,pur_resale_fa,oth_cur_assets,total_cur_assets,fa_avail_for_sale,htm_invest,lt_eqt_invest,invest_real_estate,time_deposits,oth_assets,lt_rec,fix_assets,cip,const_materials,fixed_assets_disp,produc_bio_assets,oil_and_gas_assets,intan_assets,r_and_d,goodwill,lt_amor_exp,defer_tax_assets,decr_in_disbur,oth_nca,total_nca,cash_reser_cb,depos_in_oth_bfi,prec_metals,deriv_assets,rr_reinsur_une_prem,rr_reinsur_outstd_cla,rr_reinsur_lins_liab,rr_reinsur_lthins_liab,refund_depos,ph_pledge_loans,refund_cap_depos,indep_acct_assets,client_depos,client_prov,transac_seat_fee,invest_as_receiv,total_assets,lt_borr,st_borr,cb_borr,depos_ib_deposits,loan_oth_bank,trading_fl,notes_payable,acct_payable,adv_receipts,sold_for_repur_fa,comm_payable,payroll_payable,taxes_payable,int_payable,div_payable,oth_payable,acc_exp,deferred_inc,st_bonds_payable,payable_to_reinsurer,rsrv_insur_cont,acting_trading_sec,acting_uw_sec,non_cur_liab_due_1y,oth_cur_liab,total_cur_liab,bond_payable,lt_payable,specific_payables,estimated_liab,defer_tax_liab,defer_inc_non_cur_liab,oth_ncl,total_ncl,depos_oth_bfi,deriv_liab,depos,agency_bus_liab,oth_liab,prem_receiv_adva,depos_received,ph_invest,reser_une_prem,reser_outstd_claims,reser_lins_liab,reser_lthins_liab,indept_acc_liab,pledge_borr,indem_payable,policy_div_payable,total_liab,treasury_share,ordin_risk_reser,forex_differ,invest_loss_unconf,minority_int,total_hldr_eqy_exc_min_int,total_hldr_eqy_inc_min_int,total_liab_hldr_eqy,lt_payroll_payable,oth_comp_income,oth_eqt_tools,oth_eqt_tools_p_shr,lending_funds,acc_receivable,st_fin_payable,payables'
|
|
)
|
|
if not balancesheet.empty:
|
|
balancesheet['fetch_time'] = datetime.now()
|
|
result['balancesheet'] = balancesheet
|
|
self.logger.debug(f" 💰 资产负债表: {len(balancesheet)}条记录")
|
|
|
|
# 现金流量表
|
|
time.sleep(0.15)
|
|
cashflow = self.pro.cashflow(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231',
|
|
fields='ts_code,ann_date,f_ann_date,end_date,comp_type,report_type,net_profit,finan_exp,c_fr_sale_sg,recp_tax_rends,n_depos_incr_fi,n_incr_loans_cb,n_inc_borr_oth_fi,prem_fr_orig_contr,n_incr_insured_dep,n_reinsur_prem,n_incr_disp_tfa,ifc_cash_incr,n_incr_disp_faas,n_incr_loans_oth_bank,n_cap_incr_repur,c_fr_oth_operate_a,c_inf_fr_operate_a,c_paid_goods_s,c_paid_to_for_empl,c_paid_for_taxes,n_incr_clt_loan_adv,n_incr_dep_cbob,c_pay_claims_orig_inco,pay_handling_chrg,pay_comm_insur_plcy,oth_cash_pay_oper_act,st_cash_out_act,n_cashflow_act,oth_recp_ral_inv_act,c_disp_withdrwl_invest,c_recp_return_invest,n_recp_disp_fiolta,n_recp_disp_sobu,stot_inflows_inv_act,c_pay_acq_const_fiolta,c_paid_invest,n_disp_subs_oth_biz,oth_pay_ral_inv_act,n_incr_pledge_loan,stot_out_inv_act,n_cashflow_inv_act,c_recp_borrow,proc_issue_bonds,oth_cash_recp_ral_fnc_act,stot_cash_in_fnc_act,free_cashflow,c_prepay_amt_borr,c_pay_dist_dpcp_int_exp,incl_dvd_profit_paid_sc_ms,oth_cashpay_ral_fnc_act,stot_cashout_fnc_act,n_cash_flows_fnc_act,eff_fx_flu_cash,n_incr_cash_cash_equ,c_cash_equ_beg_period,c_cash_equ_end_period,c_recp_cap_contrib,incl_cash_rec_saims,uncon_invest_loss,prov_depr_assets,depr_fa_coga_dpba,amort_intang_assets,lt_amort_deferred_exp,decr_deferred_exp,incr_acc_exp,loss_disp_fiolta,loss_scr_fa,loss_fv_chg,invest_loss,decr_def_inc_tax_assets,incr_def_inc_tax_liab,decr_inventories,decr_oper_payable,incr_oper_payable,others,im_net_cashflow_oper_act,conv_debt_into_cap,conv_copbonds_due_within_1y,fa_fnc_leases,end_bal_cash,beg_bal_cash,end_bal_cash_equ,beg_bal_cash_equ,im_n_incr_cash_equ'
|
|
)
|
|
if not cashflow.empty:
|
|
cashflow['fetch_time'] = datetime.now()
|
|
result['cashflow'] = cashflow
|
|
self.logger.debug(f" 💰 现金流量表: {len(cashflow)}条记录")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取财务数据失败: {e}")
|
|
|
|
return result
|
|
|
|
def fetch_financial_indicators(self, ts_code: str) -> Optional[pd.DataFrame]:
|
|
"""获取财务指标"""
|
|
try:
|
|
time.sleep(0.1)
|
|
|
|
data = self.pro.fina_indicator(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231',
|
|
fields='ts_code,ann_date,end_date,eps,dt_eps,total_revenue_ps,revenue_ps,capital_rese_ps,surplus_rese_ps,undist_profit_ps,extra_item,profit_dedt,gross_margin,current_ratio,quick_ratio,cash_ratio,invturn_days,arturn_days,inv_turn,ar_turn,ca_turn,fa_turn,assets_turn,op_income,valuechange_income,interst_income,daa,ebit,ebitda,fcff,fcfe,current_exint,noncurrent_exint,interestdebt,netdebt,tangible_asset,working_capital,networking_capital,invest_capital,retained_earnings,diluted2_eps,bps,ocfps,retainedps,cfps,ebit_ps,fcff_ps,fcfe_ps,netprofit_margin,grossprofit_margin,cogs_of_sales,expense_of_sales,profit_to_gr,saleexp_to_gr,adminexp_of_gr,finaexp_of_gr,impai_ttm,gc_of_gr,op_of_gr,ebit_of_gr,roe,roe_waa,roe_dt,roa,npta,roic,roe_yearly,roa2_yearly,roe_avg,opincome_of_ebt,investincome_of_ebt,n_op_profit_of_ebt,tax_to_ebt,dtprofit_to_profit,salescash_to_or,ocf_to_or,ocf_to_opincome,capitalized_to_da,debt_to_assets,assets_to_eqt,dp_assets_to_eqt,ca_to_assets,nca_to_assets,tbassets_to_totalassets,int_to_talcap,eqt_to_talcapital,currentdebt_to_debt,longdeb_to_debt,ocf_to_shortdebt,debt_to_eqt,eqt_to_debt,eqt_to_interestdebt,tangibleasset_to_debt,tangasset_to_intdebt,tangibleasset_to_netdebt,ocf_to_debt,ocf_to_interestdebt,ocf_to_netdebt,ebit_to_interest,longdebt_to_workingcapital,ebitda_to_debt,turn_days,roa_yearly,roa_dp,fixed_assets,profit_prefin_exp,non_op_profit,op_to_ebt,nop_to_ebt,ocf_to_profit,cash_to_liqdebt,cash_to_liqdebt_withinterest,op_to_liqdebt,op_to_debt,roic_yearly,total_fa_trun,profit_to_op,q_opincome,q_investincome,q_dtprofit,q_eps,q_netprofit_margin,q_gsprofit_margin,q_exp_to_sales,q_profit_to_gr,q_saleexp_to_gr,q_adminexp_to_gr,q_finaexp_to_gr,q_impair_to_gr_ttm,q_gc_to_gr,q_op_to_gr,q_roe,q_dt_roe,q_npta,q_opincome_to_ebt,q_investincome_to_ebt,q_dtprofit_to_profit,q_salescash_to_or,q_ocf_to_sales,q_ocf_to_or,basic_eps_yoy,dt_eps_yoy,cfps_yoy,op_yoy,ebt_yoy,netprofit_yoy,dt_netprofit_yoy,ocf_yoy,roe_yoy,bps_yoy,assets_yoy,eqt_yoy,tr_yoy,or_yoy,q_gr_yoy,q_gr_qoq,q_sales_yoy,q_sales_qoq,q_op_yoy,q_op_qoq,q_profit_yoy,q_profit_qoq,q_netprofit_yoy,q_netprofit_qoq,equity_yoy,rd_exp,equity_debt'
|
|
)
|
|
|
|
if not data.empty:
|
|
data['fetch_time'] = datetime.now()
|
|
self.logger.debug(f" 📊 财务指标: {len(data)}条记录")
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取财务指标失败: {e}")
|
|
return None
|
|
|
|
def fetch_company_info(self, ts_code: str) -> Dict[str, pd.DataFrame]:
|
|
"""获取公司基本信息"""
|
|
result = {}
|
|
|
|
try:
|
|
# 公司基本信息
|
|
time.sleep(0.1)
|
|
company = self.pro.stock_company(
|
|
ts_code=ts_code,
|
|
fields='ts_code,exchange,chairman,manager,secretary,reg_capital,setup_date,province,city,introduction,website,email,office,employees,main_business,business_scope'
|
|
)
|
|
if not company.empty:
|
|
company['fetch_time'] = datetime.now()
|
|
result['company'] = company
|
|
self.logger.debug(f" 🏢 公司信息: {len(company)}条记录")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取公司信息失败: {e}")
|
|
|
|
return result
|
|
|
|
def fetch_shareholder_data(self, ts_code: str) -> Dict[str, pd.DataFrame]:
|
|
"""获取股东数据"""
|
|
result = {}
|
|
|
|
try:
|
|
# 前十大股东
|
|
time.sleep(0.15)
|
|
top10_holders = self.pro.top10_holders(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231'
|
|
)
|
|
if not top10_holders.empty:
|
|
top10_holders['fetch_time'] = datetime.now()
|
|
result['top10_holders'] = top10_holders
|
|
self.logger.debug(f" 👥 前十大股东: {len(top10_holders)}条记录")
|
|
|
|
# 前十大流通股东
|
|
time.sleep(0.15)
|
|
top10_floatholders = self.pro.top10_floatholders(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231'
|
|
)
|
|
if not top10_floatholders.empty:
|
|
top10_floatholders['fetch_time'] = datetime.now()
|
|
result['top10_floatholders'] = top10_floatholders
|
|
self.logger.debug(f" 👥 前十大流通股东: {len(top10_floatholders)}条记录")
|
|
|
|
# 股东人数
|
|
time.sleep(0.1)
|
|
stk_holdernumber = self.pro.stk_holdernumber(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231'
|
|
)
|
|
if not stk_holdernumber.empty:
|
|
stk_holdernumber['fetch_time'] = datetime.now()
|
|
result['stk_holdernumber'] = stk_holdernumber
|
|
self.logger.debug(f" 👥 股东人数: {len(stk_holdernumber)}条记录")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取股东数据失败: {e}")
|
|
|
|
return result
|
|
|
|
def fetch_fund_data(self, ts_code: str) -> Dict[str, pd.DataFrame]:
|
|
"""获取基金持股数据"""
|
|
result = {}
|
|
|
|
try:
|
|
# 基金持股
|
|
time.sleep(0.15)
|
|
fund_holdings = self.pro.fund_holdings(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date[:4] + '0101',
|
|
end_date=self.end_date[:4] + '1231'
|
|
)
|
|
if not fund_holdings.empty:
|
|
fund_holdings['fetch_time'] = datetime.now()
|
|
result['fund_holdings'] = fund_holdings
|
|
self.logger.debug(f" 🏦 基金持股: {len(fund_holdings)}条记录")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取基金数据失败: {e}")
|
|
|
|
return result
|
|
|
|
def fetch_financing_data(self, ts_code: str) -> Dict[str, pd.DataFrame]:
|
|
"""获取融资融券数据"""
|
|
result = {}
|
|
|
|
try:
|
|
# 融资融券交易汇总
|
|
time.sleep(0.1)
|
|
margin = self.pro.margin(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date,
|
|
end_date=self.end_date
|
|
)
|
|
if not margin.empty:
|
|
margin['fetch_time'] = datetime.now()
|
|
result['margin'] = margin
|
|
self.logger.debug(f" 💳 融资融券: {len(margin)}条记录")
|
|
|
|
# 融资融券交易明细
|
|
time.sleep(0.1)
|
|
margin_detail = self.pro.margin_detail(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date,
|
|
end_date=self.end_date
|
|
)
|
|
if not margin_detail.empty:
|
|
margin_detail['fetch_time'] = datetime.now()
|
|
result['margin_detail'] = margin_detail
|
|
self.logger.debug(f" 💳 融资融券明细: {len(margin_detail)}条记录")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取融资数据失败: {e}")
|
|
|
|
return result
|
|
|
|
def fetch_dividend_data(self, ts_code: str) -> Dict[str, pd.DataFrame]:
|
|
"""获取分红配股数据"""
|
|
result = {}
|
|
|
|
try:
|
|
# 分红送股
|
|
time.sleep(0.1)
|
|
dividend = self.pro.dividend(
|
|
ts_code=ts_code,
|
|
fields='ts_code,end_date,ann_date,div_proc,stk_div,stk_bo_rate,stk_co_rate,cash_div,cash_div_tax,record_date,ex_date,pay_date,div_listdate,imp_ann_date'
|
|
)
|
|
if not dividend.empty:
|
|
dividend['fetch_time'] = datetime.now()
|
|
result['dividend'] = dividend
|
|
self.logger.debug(f" 💰 分红配股: {len(dividend)}条记录")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取分红数据失败: {e}")
|
|
|
|
return result
|
|
|
|
def fetch_announcement_data(self, ts_code: str) -> Optional[pd.DataFrame]:
|
|
"""获取公告数据"""
|
|
try:
|
|
time.sleep(0.1)
|
|
|
|
data = self.pro.anns(
|
|
ts_code=ts_code,
|
|
start_date=self.start_date,
|
|
end_date=self.end_date,
|
|
fields='ts_code,ann_date,ann_type,title,content'
|
|
)
|
|
|
|
if not data.empty:
|
|
data['fetch_time'] = datetime.now()
|
|
self.logger.debug(f" 📢 公告: {len(data)}条记录")
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 获取公告数据失败: {e}")
|
|
return None
|
|
|
|
def fetch_technical_indicators(self, daily_data: pd.DataFrame) -> pd.DataFrame:
|
|
"""计算技术指标"""
|
|
if daily_data.empty:
|
|
return daily_data
|
|
|
|
try:
|
|
# 确保数据按日期排序
|
|
daily_data = daily_data.sort_values('trade_date')
|
|
|
|
# 移动平均线
|
|
daily_data['ma5'] = daily_data['close'].rolling(window=5).mean()
|
|
daily_data['ma10'] = daily_data['close'].rolling(window=10).mean()
|
|
daily_data['ma20'] = daily_data['close'].rolling(window=20).mean()
|
|
daily_data['ma60'] = daily_data['close'].rolling(window=60).mean()
|
|
|
|
# RSI
|
|
delta = daily_data['close'].diff()
|
|
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
|
|
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
|
|
rs = gain / loss
|
|
daily_data['rsi'] = 100 - (100 / (1 + rs))
|
|
|
|
# MACD
|
|
ema12 = daily_data['close'].ewm(span=12).mean()
|
|
ema26 = daily_data['close'].ewm(span=26).mean()
|
|
daily_data['macd'] = ema12 - ema26
|
|
daily_data['macd_signal'] = daily_data['macd'].ewm(span=9).mean()
|
|
daily_data['macd_hist'] = daily_data['macd'] - daily_data['macd_signal']
|
|
|
|
# 布林带
|
|
daily_data['bb_middle'] = daily_data['close'].rolling(window=20).mean()
|
|
bb_std = daily_data['close'].rolling(window=20).std()
|
|
daily_data['bb_upper'] = daily_data['bb_middle'] + (bb_std * 2)
|
|
daily_data['bb_lower'] = daily_data['bb_middle'] - (bb_std * 2)
|
|
|
|
# KDJ
|
|
low_min = daily_data['low'].rolling(window=9).min()
|
|
high_max = daily_data['high'].rolling(window=9).max()
|
|
rsv = (daily_data['close'] - low_min) / (high_max - low_min) * 100
|
|
daily_data['kdj_k'] = rsv.ewm(com=2).mean()
|
|
daily_data['kdj_d'] = daily_data['kdj_k'].ewm(com=2).mean()
|
|
daily_data['kdj_j'] = 3 * daily_data['kdj_k'] - 2 * daily_data['kdj_d']
|
|
|
|
self.logger.debug(f" 📊 技术指标计算完成")
|
|
return daily_data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 技术指标计算失败: {e}")
|
|
return daily_data
|
|
|
|
def save_to_database(self, data_dict: Dict, ts_code: str):
|
|
"""保存数据到数据库"""
|
|
try:
|
|
conn = pymysql.connect(**self.db_config)
|
|
|
|
for category, data in data_dict.items():
|
|
if data is None or (isinstance(data, pd.DataFrame) and data.empty):
|
|
continue
|
|
|
|
# 处理不同类型的数据
|
|
if isinstance(data, dict):
|
|
# 嵌套字典，如财务数据
|
|
for sub_category, sub_data in data.items():
|
|
if isinstance(sub_data, pd.DataFrame) and not sub_data.empty:
|
|
table_name = f"{category}_{sub_category}"
|
|
self._save_dataframe_to_mysql(conn, sub_data, table_name)
|
|
elif isinstance(data, pd.DataFrame):
|
|
# 直接是DataFrame
|
|
table_name = category
|
|
self._save_dataframe_to_mysql(conn, data, table_name)
|
|
|
|
conn.close()
|
|
self.logger.debug(f" 💾 数据保存成功")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" ❌ 数据保存失败: {e}")
|
|
|
|
def _save_dataframe_to_mysql(self, conn, df: pd.DataFrame, table_name: str):
|
|
"""保存DataFrame到MySQL"""
|
|
try:
|
|
# 转换数据类型
|
|
df = df.copy()
|
|
|
|
# 处理时间戳类型
|
|
for col in df.columns:
|
|
if df[col].dtype == 'object':
|
|
try:
|
|
df[col] = pd.to_datetime(df[col], errors='ignore')
|
|
except:
|
|
pass
|
|
|
|
# 使用to_sql保存数据
|
|
df.to_sql(
|
|
name=table_name,
|
|
con=conn,
|
|
if_exists='append',
|
|
index=False,
|
|
method='multi',
|
|
chunksize=1000
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.debug(f" ⚠️ 表 {table_name} 保存警告: {e}")
|
|
|
|
def fetch_stock_comprehensive_data(self, ts_code: str, stock_name: str) -> bool:
|
|
"""获取单只股票的全面数据"""
|
|
self.logger.info(f"📈 处理股票: {stock_name}({ts_code})")
|
|
|
|
all_data = {}
|
|
|
|
try:
|
|
# 1. 日线数据
|
|
daily_data = self.fetch_daily_data(ts_code)
|
|
if daily_data is not None:
|
|
# 计算技术指标
|
|
daily_data = self.fetch_technical_indicators(daily_data)
|
|
all_data['daily'] = daily_data
|
|
|
|
# 2. 周线和月线数据
|
|
weekly_monthly = self.fetch_weekly_monthly_data(ts_code)
|
|
all_data.update(weekly_monthly)
|
|
|
|
# 3. 财务数据
|
|
financial_data = self.fetch_financial_data(ts_code)
|
|
all_data['financial'] = financial_data
|
|
|
|
# 4. 财务指标
|
|
financial_indicators = self.fetch_financial_indicators(ts_code)
|
|
if financial_indicators is not None:
|
|
all_data['financial_indicators'] = financial_indicators
|
|
|
|
# 5. 公司基本信息
|
|
company_info = self.fetch_company_info(ts_code)
|
|
all_data.update(company_info)
|
|
|
|
# 6. 股东数据
|
|
shareholder_data = self.fetch_shareholder_data(ts_code)
|
|
all_data.update(shareholder_data)
|
|
|
|
# 7. 基金持股
|
|
fund_data = self.fetch_fund_data(ts_code)
|
|
all_data.update(fund_data)
|
|
|
|
# 8. 融资融券
|
|
financing_data = self.fetch_financing_data(ts_code)
|
|
all_data.update(financing_data)
|
|
|
|
# 9. 分红配股
|
|
dividend_data = self.fetch_dividend_data(ts_code)
|
|
all_data.update(dividend_data)
|
|
|
|
# 10. 公告数据 (限制数量，避免过多数据)
|
|
announcement_data = self.fetch_announcement_data(ts_code)
|
|
if announcement_data is not None:
|
|
all_data['announcements'] = announcement_data
|
|
|
|
# 保存到数据库
|
|
self.save_to_database(all_data, ts_code)
|
|
|
|
self.logger.info(f"✅ {stock_name}({ts_code}) 数据获取完成")
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"❌ {stock_name}({ts_code}) 数据获取失败: {e}")
|
|
return False
|
|
|
|
def run_comprehensive_fetch(self):
|
|
"""运行全面数据获取"""
|
|
self.logger.info("🚀 开始5年历史数据全面获取...")
|
|
|
|
# 获取股票列表
|
|
stocks_df = self.get_stock_list()
|
|
if stocks_df.empty:
|
|
self.logger.error("❌ 无法获取股票列表，退出")
|
|
return
|
|
|
|
total_stocks = len(stocks_df)
|
|
self.progress['total_stocks'] = total_stocks
|
|
|
|
# 初始化进度
|
|
processed_count = 0
|
|
success_count = 0
|
|
failed_stocks = []
|
|
|
|
self.logger.info(f"📋 开始处理 {total_stocks} 只股票")
|
|
|
|
# 使用线程池并行处理
|
|
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
|
|
# 提交所有任务
|
|
future_to_stock = {}
|
|
|
|
for idx, row in stocks_df.iterrows():
|
|
ts_code = row['ts_code']
|
|
stock_name = row['name']
|
|
|
|
# 检查是否已处理
|
|
if ts_code in self.progress.get('completed_stocks', []):
|
|
processed_count += 1
|
|
continue
|
|
|
|
future = executor.submit(self.fetch_stock_comprehensive_data, ts_code, stock_name)
|
|
future_to_stock[future] = (ts_code, stock_name)
|
|
|
|
# 控制并发数
|
|
if len(future_to_stock) >= self.max_workers:
|
|
# 等待一批完成
|
|
for future in as_completed(list(future_to_stock.keys())[:5]):
|
|
ts_code, stock_name = future_to_stock[future]
|
|
try:
|
|
success = future.result()
|
|
processed_count += 1
|
|
|
|
if success:
|
|
success_count += 1
|
|
# 记录已完成的股票
|
|
if 'completed_stocks' not in self.progress:
|
|
self.progress['completed_stocks'] = []
|
|
self.progress['completed_stocks'].append(ts_code)
|
|
else:
|
|
failed_stocks.append((ts_code, stock_name))
|
|
|
|
# 更新进度
|
|
self.progress['processed_stocks'] = processed_count
|
|
|
|
# 定期保存进度
|
|
if processed_count % 50 == 0:
|
|
self.save_progress()
|
|
self.logger.info(f"📊 进度: {processed_count}/{total_stocks} ({processed_count/total_stocks*100:.1f}%)")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"❌ 股票 {stock_name}({ts_code}) 处理异常: {e}")
|
|
failed_stocks.append((ts_code, stock_name))
|
|
|
|
del future_to_stock[future]
|
|
|
|
# 处理剩余任务
|
|
for future in as_completed(future_to_stock.keys()):
|
|
ts_code, stock_name = future_to_stock[future]
|
|
try:
|
|
success = future.result()
|
|
processed_count += 1
|
|
|
|
if success:
|
|
success_count += 1
|
|
if 'completed_stocks' not in self.progress:
|
|
self.progress['completed_stocks'] = []
|
|
self.progress['completed_stocks'].append(ts_code)
|
|
else:
|
|
failed_stocks.append((ts_code, stock_name))
|
|
|
|
self.progress['processed_stocks'] = processed_count
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"❌ 股票 {stock_name}({ts_code}) 处理异常: {e}")
|
|
failed_stocks.append((ts_code, stock_name))
|
|
|
|
# 保存最终进度
|
|
self.save_progress()
|
|
|
|
# 生成报告
|
|
self.generate_completion_report(total_stocks, success_count, failed_stocks)
|
|
|
|
def generate_completion_report(self, total_stocks: int, success_count: int, failed_stocks: List[Tuple[str, str]]):
|
|
"""生成完成报告"""
|
|
report = {
|
|
'completion_time': datetime.now().isoformat(),
|
|
'total_stocks': total_stocks,
|
|
'success_count': success_count,
|
|
'failed_count': len(failed_stocks),
|
|
'success_rate': success_count / total_stocks * 100 if total_stocks > 0 else 0,
|
|
'failed_stocks': failed_stocks,
|
|
'time_range': f"{self.start_date} ~ {self.end_date}",
|
|
'data_categories': [
|
|
'日线数据', '周线数据', '月线数据', '技术指标',
|
|
'财务数据(利润表/资产负债表/现金流量表)', '财务指标',
|
|
'公司基本信息', '股东结构', '基金持股',
|
|
'融资融券', '分红配股', '公告数据'
|
|
]
|
|
}
|
|
|
|
# 保存报告
|
|
report_file = f'logs/comprehensive_5year_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json'
|
|
with open(report_file, 'w', encoding='utf-8') as f:
|
|
json.dump(report, f, ensure_ascii=False, indent=2)
|
|
|
|
# 打印摘要
|
|
self.logger.info("🎉 数据获取完成！")
|
|
self.logger.info(f"📊 总计股票: {total_stocks}")
|
|
self.logger.info(f"✅ 成功: {success_count} ({success_count/total_stocks*100:.1f}%)")
|
|
self.logger.info(f"❌ 失败: {len(failed_stocks)} ({len(failed_stocks)/total_stocks*100:.1f}%)")
|
|
self.logger.info(f"📄 详细报告: {report_file}")
|
|
|
|
if failed_stocks:
|
|
self.logger.info("❌ 失败股票列表:")
|
|
for ts_code, stock_name in failed_stocks[:10]: # 只显示前10个
|
|
self.logger.info(f" - {stock_name}({ts_code})")
|
|
if len(failed_stocks) > 10:
|
|
self.logger.info(f" ... 还有 {len(failed_stocks)-10} 只股票")
|
|
|
|
def main():
|
|
"""主函数"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description='TuShare 5年历史数据全面获取器')
|
|
parser.add_argument('--token', type=str, help='TuShare Pro API Token')
|
|
parser.add_argument('--test', action='store_true', help='测试模式 (只处理前5只股票)')
|
|
parser.add_argument('--incremental', action='store_true', help='增量更新模式')
|
|
parser.add_argument('--full', action='store_true', help='全量更新模式')
|
|
parser.add_argument('--deep-analysis', action='store_true', help='深度分析模式')
|
|
parser.add_argument('--start-date', type=str, help='开始日期 YYYYMMDD')
|
|
parser.add_argument('--end-date', type=str, help='结束日期 YYYYMMDD')
|
|
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
fetcher = ComprehensiveDataFetcher(token=args.token)
|
|
|
|
# 根据参数调整时间范围
|
|
if args.start_date:
|
|
fetcher.start_date = args.start_date
|
|
if args.end_date:
|
|
fetcher.end_date = args.end_date
|
|
|
|
if args.test:
|
|
# 测试模式
|
|
fetcher.logger.info("🧪 测试模式 - 只处理前5只股票")
|
|
stocks_df = fetcher.get_stock_list()
|
|
if not stocks_df.empty:
|
|
test_stocks = stocks_df.head(5)
|
|
for idx, row in test_stocks.iterrows():
|
|
ts_code = row['ts_code']
|
|
stock_name = row['name']
|
|
fetcher.fetch_stock_comprehensive_data(ts_code, stock_name)
|
|
fetcher.logger.info("🧪 测试模式完成")
|
|
|
|
elif args.incremental:
|
|
# 增量更新模式
|
|
fetcher.logger.info("📊 增量更新模式")
|
|
if not args.start_date:
|
|
# 默认获取昨天的数据
|
|
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d')
|
|
fetcher.start_date = yesterday
|
|
fetcher.end_date = yesterday
|
|
fetcher.run_comprehensive_fetch()
|
|
|
|
elif args.full:
|
|
# 全量更新模式
|
|
fetcher.logger.info("🔄 全量更新模式")
|
|
fetcher.run_comprehensive_fetch()
|
|
|
|
elif args.deep_analysis:
|
|
# 深度分析模式
|
|
fetcher.logger.info("🧠 深度分析模式")
|
|
# 可以在这里添加额外的分析逻辑
|
|
fetcher.run_comprehensive_fetch()
|
|
|
|
else:
|
|
# 默认模式
|
|
fetcher.run_comprehensive_fetch()
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n⚠️ 用户中断操作")
|
|
except Exception as e:
|
|
print(f"❌ 运行失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
if __name__ == "__main__":
|
|
main() |