fin-data-acquisition
根据REFINED_DESIGN.md中的变量定义，自动获取所需数据并生成可执行的回归分析脚本（Python/Stata）。
触发条件
- 关键词:
获取数据数据获取data acquisition下载数据数据准备实证数据 - Skill语法:
Skill: fin-data-acquisition - 前置条件: 已完成
REFINED_DESIGN.md(研究设计文档)
核心原则
禁止行为 (未经用户明确授权不得执行)
❌ 静默回退到模拟数据 ❌ 自动生成虚假回归结果 ❌ 在用户未同意情况下继续流水线使用模拟数据 ❌ 跳过数据源预检查直接获取数据
数据源预检查 (强制执行)
在任何数据获取前，必须先运行数据源检查：
pythonfrom scripts.data_source_checker import DataSourceChecker, DataRequirement # 第一步：定义数据需求 requirements = [ DataRequirement( name="financial_data", user_facing_name="A股财务数据", description="ROA、资产负债率、企业规模、研发投入", sources=["tushare", "wind", "csmar", "akshare"], required=True, ), DataRequirement( name="esg_data", user_facing_name="ESG评级", sources=["msci", "商道融绿", "华证"], required=False, ), DataRequirement( name="macro_data", user_facing_name="宏观数据", sources=["user-financial", "user-wb-data", "user-imf-data"], required=True, ), ] # 第二步：运行数据源检查 checker = DataSourceChecker() results = checker.check(requirements) # 第三步：展示可用性报告 checker.print_report(results)
数据源Fallback链
每个数据类型都有明确的降级路径：
A股财务数据
tushare (需TUSHARE_TOKEN) ↓ 失败/无Token wind (需Wind账号) ↓ 失败/无账号 csmar (需机构账号) ↓ 失败/无账号 akshare (免费，备选) ↓ 失败 手动下载 -> 询问用户
宏观数据
user-financial (akshare, 免费) ↓ 失败 user-wb-data (World Bank, 免费) ↓ 失败 user-imf-data (IMF, 免费) ↓ 失败 手动下载 -> 询问用户
美股数据
user-yfinance (免费) ↓ 失败 user-eodhd (需EODHD_API_KEY) ↓ 失败/无Key 手动下载 -> 询问用户
学术文献数据
user-openalex (免费) ↓ 失败 user-arxiv (免费) ↓ 失败 user-nber-wp (免费) ↓ 失败 手动检索 -> 询问用户
DataFetcher API
pythonfrom scripts.research_framework import DataFetcher, ProvenanceTracker # 初始化 (带数据溯源) tracker = ProvenanceTracker(output_dir="data/provenance/") fetcher = DataFetcher(output_dir="data/", tracker=tracker, verbose=True) # ============ 面板数据获取 ============ df = fetcher.fetch_panel( tickers=["000001.SZ", "600000.SH"], years=["2018", "2019", "2020", "2021", "2022"], statements=["balance", "income", "cashflow"], include_sustainability=True, # ESG数据 ) # ============ 财务报表获取 ============ fin = fetcher.fetch_financials("000001.SZ", "income") # 利润表 fin = fetcher.fetch_financials("000001.SZ", "balance") # 资产负债表 fin = fetcher.fetch_financials("000001.SZ", "cashflow") # 现金流量表 # ============ 公司信息获取 ============ info = fetcher.fetch_ticker_info("000001.SZ") # 股票基本信息 # ============ ESG/可持续发展数据 ============ sust = fetcher.fetch_sustainability("000001.SZ") # ESG评级等 # ============ 宏观数据获取 ============ macro = fetcher.fetch_macro(indicator="gdp", country="CHN") # ============ 融资融券数据 ============ margin = fetcher.fetch_margin(ts_code="000001.SZ", start_date="20180101") # ============ 陆股通/港股通 ============ hgt = fetcher.fetch_hsgt_top10(date="20240101") # ============ 分析师预测 ============ forecast = fetcher.fetch_consensus("000001.SZ")
MCP数据获取
直接调用MCP工具
python# A股行情数据 (tushare) server: user-tushare tool: get_daily_quote params: { "ts_code": "000001.SZ", "start_date": "20240101", "end_date": "20241231" } # 财务报告 (tushare) server: user-tushare tool: get_financial_report params: { "ts_code": "000001.SZ", "report_type": "income" } # 美股数据 (yfinance) server: user-yfinance tool: get_yf_historical params: { "ticker": "AAPL", "start_date": "2024-01-01", "end_date": "2024-12-31" } # 宏观数据 (World Bank) server: user-wb-data tool: get_wb_indicator params: { "country_code": "CHN", "indicator": "wb_gdp_usd" } # 研报数据 (eastmoney) server: user-eastmoney-reports tool: get_research_report params: { "ts_code": "000001.SZ", "max_results": 20 }
回归脚本生成
Python脚本模板
python""" {研究标题} — 回归分析脚本 Generated by fin-data-acquisition skill Date: {date} """ import pandas as pd import numpy as np import statsmodels.api as sm from scipy import stats import matplotlib.pyplot as plt import seaborn as sns import warnings warnings.filterwarnings('ignore') # ============ 配置 ============ DATA_PATH = "data/processed/{dataset_name}.csv" OUTPUT_DIR = "output/fin-experiments/" # ============ 数据加载 ============ df = pd.read_csv(DATA_PATH) print(f"样本量: {len(df)}, 时间范围: {df['year'].min()}-{df['year'].max()}") print(f"处理组: {df['treat'].sum()}, 对照组: {(1-df['treat']).sum()}") # ============ 描述性统计 ============ desc = df[['y_var', 'x_var', 'controls']].describe() desc.to_csv(f"{OUTPUT_DIR}/descriptive_stats.csv") print(desc) # ============ 相关性矩阵 ============ corr = df[['y_var', 'x_var', 'controls']].corr() sns.heatmap(corr, annot=True, cmap='RdBu_r', center=0) plt.savefig(f"{OUTPUT_DIR}/correlation_matrix.pdf", dpi=300) plt.close() # ============ 基准回归 ============ # OLS X = sm.add_constant(df[['x_var'] + ['controls']]) y = df['y_var'] model = sm.OLS(y, X).fit(cov_type='cluster', cov_kwds={'groups': df['firmid']}) print(model.summary()) # DID回归 (带双向固定效应) from linearmodels.panel import PanelOLS df = df.set_index(['firmid', 'year']) mod = PanelOLS.from_formula('y_var ~ x_var + EntityEffects + TimeEffects', df) res = mod.fit(cov_type='clustered', cluster_entity=True) print(res.summary) # ============ 保存结果 ============ results = { 'coefficient': model.params, 'std_error': model.bse, 'p_value': model.pvalues, 'r_squared': model.rsquared, } pd.DataFrame(results).to_csv(f"{OUTPUT_DIR}/regression_results.csv")
Stata脚本模板
stata/* {研究标题} — Stata回归分析 Generated by fin-data-acquisition skill Date: {date} */ clear all cd "data/processed/" * 加载数据 import delimited "{dataset_name}.csv", clear * 描述性统计 estpost summarize y_var x_var controls, detail esttab using "descriptive_stats.tex", cells("mean sd min p25 p50 p75 max") replace * 相关性矩阵 pwcorr y_var x_var controls, star(5) sig * 基准回归 (OLS + 聚类标准误) reg y_var x_var controls, vce(cluster firmid) * 双向固定效应 (DID) encode firmid, gen(firm) encode year, gen(year_dum) xtset firm year_dum xtreg y_var x_var controls i.year_dum, fe vce(cluster firm) * 平行趋势检验 gen pre1 = (year == treated_year - 1) gen post0 = (year == treated_year) gen post1 = (year == treated_year + 1) reg y_var pre1 post0 post1 controls i.year_dum, vce(cluster firm) * 安慰剂检验 xtreg y_var placebo_* controls i.year_dum, fe vce(cluster firm) * 异质性分析 bysort group: xtreg y_var x_var controls i.year_dum, fe vce(cluster firm) esttab using "regression_results.tex", b(4) se(4) star(* 0.1 ** 0.05 *** 0.01) replace
数据溯源追踪
每个数据获取操作都记录溯源信息：
pythonfrom scripts.core.provenance import DataProvenance provenance = DataProvenance() # 记录数据获取 provenance.record_fetch( source="tushare", table="financial_statement", timestamp=datetime.now(), rows=len(df), fields=list(df.columns), query_params={"ts_code": "000001.SZ", "year": 2022}, ) # 记录数据转换 provenance.record_transform( input_tables=["financial_statement", "trading_data"], output_table="merged_panel", transformation="merge on [ts_code, year]", rows_before=[1000, 500], rows_after=800, ) # 导出溯源报告 provenance.export("data/provenance/provenance_report.json")
Checkpoint (强制交互)
[CHECKPOINT] 数据源预检查完成。 可用性报告: ✅ A股财务数据: tushare 可用 (Token已配置) ✅ 宏观数据: user-financial 可用 ⚠️ ESG数据: MSCI需账号 (可选数据) ❌ 陆股通数据: tushare当前版本不支持 问题数据: - 陆股通成分股数据需要Wind账号或手动下载 请选择: 1. 授权使用模拟数据 (用于测试脚本) 2. 提供替代数据源 3. 更换研究变量/设计 4. 继续 (缺失数据将在后续处理)
依赖项
scripts/data_source_checker.py— 数据源预检查scripts/research_framework/data_fetcher.py— 数据获取接口scripts/core/provenance.py— 数据溯源追踪scripts/research_framework/regression_engine.py— 回归引擎
约束
- 必须先运行数据源预检查 — 不跳过他
- 禁止静默Fallback — 模拟数据必须用户授权
- 每个数据操作必须记录溯源 — 包括来源、时间戳、行数
- 检查点强制暂停 — 用户确认前不继续
- 失败时显示具体原因 — 而非笼统错误
