fin-data-acquisition v1.0.0A

Immutable. This exact content is served forever at /api/v1/blob/03bd0a8ffd183874.

---
name: fin-data-acquisition
description: 根据REFINED_DESIGN.md中的变量定义，自动获取所需数据并生成可执行的回归分析脚本（Python/Stata）。
trigger: "获取数据|数据获取|data acquisition|下载数据|数据准备"
version: 1.0.0
created: 2026-06-13
tags: [data, acquisition, mcp, python, stata, regression]
---

# fin-data-acquisition

根据REFINED_DESIGN.md中的变量定义，自动获取所需数据并生成可执行的回归分析脚本（Python/Stata）。

## 触发条件

- 关键词: `获取数据` `数据获取` `data acquisition` `下载数据` `数据准备` `实证数据`
- Skill语法: `Skill: fin-data-acquisition`
- 前置条件: 已完成 `REFINED_DESIGN.md` (研究设计文档)

## 核心原则

### 禁止行为 (未经用户明确授权不得执行)

```
❌ 静默回退到模拟数据
❌ 自动生成虚假回归结果
❌ 在用户未同意情况下继续流水线使用模拟数据
❌ 跳过数据源预检查直接获取数据
```

### 数据源预检查 (强制执行)

在任何数据获取前，必须先运行数据源检查：

```python
from 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

```python
from 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
```

## 数据溯源追踪

每个数据获取操作都记录溯源信息：

```python
from 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` — 回归引擎

## 约束

1. **必须先运行数据源预检查** — 不跳过他
2. **禁止静默Fallback** — 模拟数据必须用户授权
3. **每个数据操作必须记录溯源** — 包括来源、时间戳、行数
4. **检查点强制暂停** — 用户确认前不继续
5. **失败时显示具体原因** — 而非笼统错误