ljwx-stock/DATA_PIPELINE_IMPLEMENTATION.md

A股数据采集与处理完整流程实施文档

Complete Data Pipeline Implementation for A-Shares

版本: 1.0 日期: 2025-11-11 状态: 🚧 实施中

---

📋 实施进度

• PostgreSQL数据库Schema设计 (12张表, 30+索引)
• Docker Compose配置 (PostgreSQL + Redis + pgAdmin)
• SQLAlchemy ORM模型定义
• TuShare数据爬虫实现
• 数据标准化模块
• 特征计算引擎
• ML预测管道
• 推荐生成系统

---

一、PostgreSQL数据库架构

1.1 核心表结构

表名	功能	记录数(预估)	关键字段
t_stock_basic	股票基础信息	5,000+	ts_code, symbol, name, industry
t_stock_daily	日线行情	1,000万+	symbol, trade_date, open, close, volume
t_stock_features	技术指标	1,000万+	symbol, trade_date, ma5, rsi, macd
t_stock_fundamental	基本面数据	10万+	symbol, trade_date, pe_ttm, roe
t_predictions	ML预测结果	10万+	symbol, score_ml, probability
t_recommendations	推荐结果	1万+	symbol, rank, score_final
t_backtest_results	回测结果	100+	strategy_id, sharpe_ratio, win_rate
t_market_sentiment	市场情绪	1,000+	trade_date, overall_sentiment
t_announcements	公司公告	10万+	symbol, title, impact_score
t_data_fetch_logs	采集日志	1,000+	task_name, status, duration

1.2 数据库特性

✅ JSONB字段应用:

• t_recommendations.reason - LLM生成的推荐理由
• t_predictions.feature_importance - 特征重要性
• t_backtest_results.equity_curve - 净值曲线
• t_announcements.extracted_info - 公告关键信息

✅ 索引优化:

-- 复合索引示例
CREATE INDEX idx_daily_symbol_date ON t_stock_daily(symbol, trade_date DESC);
CREATE INDEX idx_features_symbol_date ON t_stock_features(symbol, trade_date DESC);

-- 性能索引
CREATE INDEX idx_predictions_score ON t_predictions(score_ml DESC);
CREATE INDEX idx_recommendations_rank ON t_recommendations(rank);

✅ 视图:

• v_latest_recommendations - 最新推荐
• v_stock_complete_info - 股票完整信息

✅ 触发器:

• 自动更新 updated_at 时间戳

---

二、Docker Compose配置

2.1 服务组成

services:
  postgres:     # PostgreSQL 16
  redis:        # Redis 7
  pgadmin:      # pgAdmin 4 (数据库管理)

2.2 连接信息

PostgreSQL:
- Host: localhost
- Port: 5432
- Database: stock_ai
- User: stockai
- Password: stockai2025

Redis:
- Host: localhost
- Port: 6379
- Password: redis2025

pgAdmin:
- URL: http://localhost:5050
- Email: admin@stockai.com
- Password: admin2025

2.3 启动命令

# 启动所有服务
cd /root/work/stock
docker-compose up -d

# 查看日志
docker-compose logs -f postgres

# 停止服务
docker-compose down

# 清理数据卷(危险!)
docker-compose down -v

---

三、数据采集流程

3.1 TuShare爬虫架构

class TuShareCrawler:
    def crawl_all_stocks(start_date, end_date, limit=None):
        """
        完整采集流程:
        1. 获取股票列表 (5000+只A股)
        2. 保存股票基础信息
        3. 逐只采集日线数据 (365天)
        4. 逐只采集基本面数据
        5. 记录采集日志
        """

3.2 采集统计

预估数据量 (一年数据):

股票数量: 5,000只
交易日: ~245天/年

日线数据: 5,000 × 245 = 1,225,000条
基本面数据: 5,000 × 4季度 = 20,000条
总数据量: ~125万条记录

采集时间预估:

单只股票耗时: ~2秒 (含限流)
总耗时: 5,000 × 2s / 60 = ~167分钟 = 2.8小时

3.3 限流策略

# TuShare API限流
request_interval = 0.3秒  # 每次请求间隔

# 批量操作
batch_size = 1000  # 批量插入数据库

3.4 错误处理

# 单只股票失败不影响整体
try:
    crawl_stock(symbol)
except Exception as e:
    log_error(symbol, e)
    continue  # 继续下一只

# 失败记录统计
stats = {
    'success_stocks': 4,950,
    'failed_stocks': 50,
    'errors': [...]
}

---

四、数据标准化流程

4.1 数据清洗

class DataStandardizer:
    def clean_daily_data(df: pd.DataFrame):
        """
        清洗步骤:
        1. 移除停牌日数据 (volume=0)
        2. 填充缺失值 (前向填充)
        3. 异常值处理 (涨跌幅>20%检查)
        4. 数据类型转换
        """

    def validate_data(df: pd.DataFrame):
        """
        数据验证:
        1. 价格合理性 (open/close/high/low)
        2. 成交量非负
        3. 日期连续性检查
        """

4.2 数据规范化

# 价格单位统一
price_columns = ['open', 'high', 'low', 'close']
df[price_columns] = df[price_columns].round(3)

# 成交量单位 (手 → 股)
df['volume'] = df['vol'] * 100

# 成交额单位 (千元 → 元)
df['amount'] = df['amount'] * 1000

# 百分比字段
df['change_pct'] = df['pct_chg']
df['turnover_rate'] = df['turnover_rate']

---

五、特征计算引擎

5.1 技术指标计算

class FeatureCalculator:
    def calculate_all_features(df: pd.DataFrame):
        """
        计算40+技术指标:

        移动平均线 (5个):
        - MA5, MA10, MA20, MA50, MA200

        动量指标 (8个):
        - RSI(6,12,24), MACD, KDJ(K,D,J), ROC

        波动率指标 (7个):
        - ATR, Bollinger Bands (4), Volatility(5d,20d)

        成交量指标 (3个):
        - Volume_MA5, Volume_MA10, Volume_Ratio

        其他指标:
        - Momentum(5d,20d), Price_Position_in_BB
        """

5.2 特征工程

# 特征组合
feature_combinations = [
    ('ma5', 'ma20'),      # 金叉死叉
    ('rsi', 'rsi_6'),     # RSI背离
    ('macd', 'macd_signal'),  # MACD交叉
    ('close', 'boll_upper')   # 布林带突破
]

# 特征标准化
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
normalized_features = scaler.fit_transform(features)

5.3 特征存储

INSERT INTO t_stock_features (
    ts_code, symbol, trade_date,
    ma5, ma10, ma20, rsi, macd, ...
) VALUES (...);

---

六、ML预测管道

6.1 模型训练

class MLPredictor:
    def train_model(features, labels):
        """
        模型选择:
        - XGBoost (主模型)
        - LightGBM (备选)

        标签定义:
        y = 1 if future_return_5d > 3% else 0

        特征选择:
        - 技术指标 (40+)
        - 基本面指标 (10+)
        - 市场情绪 (5+)
        """

        model = XGBClassifier(
            n_estimators=100,
            max_depth=6,
            learning_rate=0.1
        )

        model.fit(features, labels)
        return model

6.2 预测评分

def predict_scores(model, features):
    """
    预测流程:
    1. 特征标准化
    2. 模型预测概率
    3. 转换为评分 (0-100)
    4. 计算置信度
    """

    probabilities = model.predict_proba(features)[:, 1]
    scores = probabilities * 100

    return {
        'score_ml': scores,
        'probability': probabilities,
        'confidence': calculate_confidence(probabilities)
    }

6.3 结果存储

INSERT INTO t_predictions (
    ts_code, symbol, trade_date,
    model_type, model_version,
    score_ml, probability,
    predicted_return_5d,
    feature_importance
) VALUES (...);

---

七、推荐生成系统

7.1 推荐策略

class RecommendationEngine:
    def generate_recommendations(predictions, date):
        """
        推荐逻辑:
        1. 筛选: score_ml > 0.6
        2. 排序: 按score_ml降序
        3. 过滤:
           - 排除ST股票
           - 排除停牌股票
           - 排除成交量过低股票
        4. Top N: 选择前10-20只
        """

        top_stocks = predictions[
            (predictions['score_ml'] > 0.6) &
            (predictions['volume'] > threshold)
        ].sort_values('score_ml', ascending=False).head(10)

        return top_stocks

7.2 LLM解释生成

async def generate_llm_explanation(stock_info):
    """
    使用Claude API生成推荐理由

    输入:
    - 股票基本信息
    - 技术指标
    - 基本面数据
    - ML评分

    输出 (JSONB):
    {
        "summary": "技术面强势，MACD金叉",
        "risk": "估值偏高，注意回调风险",
        "suggestion": "建议分批建仓",
        "key_factors": ["MA金叉", "RSI超买", "成交量放大"]
    }
    """

    prompt = build_analysis_prompt(stock_info)
    response = await claude_api.generate(prompt)

    return parse_llm_response(response)

7.3 推荐存储

INSERT INTO t_recommendations (
    ts_code, symbol, name, trade_date,
    score_ml, score_llm, score_final,
    rank, reason, signal_strength,
    risk_level, expected_return
) VALUES (...);

---

八、完整数据流

┌─────────────────────────────────────────────────────────────┐
│                    1. 数据采集 (TuShare)                      │
│  ├─ 股票列表 (5000+只)                                        │
│  ├─ 日线数据 (365天 × 5000)                                   │
│  └─ 基本面数据 (季度)                                         │
└────────────────┬────────────────────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────────────────────┐
│                    2. 数据标准化                              │
│  ├─ 数据清洗 (缺失值、异常值)                                  │
│  ├─ 数据验证 (合理性检查)                                      │
│  └─ 格式统一 (单位、类型)                                      │
└────────────────┬────────────────────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────────────────────┐
│                    3. 特征计算                                │
│  ├─ 技术指标 (MA, RSI, MACD...)                              │
│  ├─ 衍生特征 (交叉信号、背离)                                  │
│  └─ 特征标准化 (StandardScaler)                               │
└────────────────┬────────────────────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────────────────────┐
│                    4. ML模型预测                              │
│  ├─ 特征输入 (40+ features)                                   │
│  ├─ 模型推理 (XGBoost)                                        │
│  └─ 评分输出 (score_ml, probability)                          │
└────────────────┬────────────────────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────────────────────┐
│                    5. 推荐生成                                │
│  ├─ 筛选排序 (Top 10-20)                                      │
│  ├─ LLM解释 (Claude API)                                      │
│  └─ 最终推荐 (rank, reason)                                   │
└────────────────┬────────────────────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────────────────────┐
│                    6. 结果存储 & API                          │
│  ├─ PostgreSQL (持久化)                                       │
│  ├─ Redis (缓存)                                              │
│  └─ FastAPI (对外服务)                                        │
└─────────────────────────────────────────────────────────────┘

---

九、使用指南

9.1 启动数据库

# 进入项目目录
cd /root/work/stock

# 启动Docker服务
docker-compose up -d

# 检查服务状态
docker-compose ps

# 查看PostgreSQL日志
docker-compose logs -f postgres

9.2 初始化数据库

from app.database_pg import init_database

# 创建所有表
init_database()

9.3 运行数据采集

from app.crawler.tushare_crawler import TuShareCrawler

# 创建爬虫实例
crawler = TuShareCrawler(token='your_tushare_token')

# 采集全部A股数据 (一年)
stats = crawler.crawl_all_stocks(
    start_date='20240101',
    end_date='20251111',
    limit=10  # 测试时限制10只股票
)

print(f"✅ Success: {stats['success_stocks']}")
print(f"📈 Daily records: {stats['total_daily_records']}")

9.4 查询数据

from app.database_pg import get_db, StockDaily

# 查询某只股票的日线数据
with get_db() as db:
    records = db.query(StockDaily).filter(
        StockDaily.symbol == '600000',
        StockDaily.trade_date >= '2024-01-01'
    ).order_by(StockDaily.trade_date.desc()).limit(100).all()

    for record in records:
        print(f"{record.trade_date}: {record.close}")

---

十、下一步工作

10.1 待实现功能

• 数据标准化模块完善
• 特征计算引擎实现
• ML模型训练管道
• 推荐生成系统
• 定时任务调度 (Celery/Airflow)
• API接口完善

10.2 性能优化

• 批量插入优化 (COPY命令)
• 分区表设计 (按月分区)
• 物化视图 (热点数据)
• 连接池优化
• 索引优化

10.3 功能扩展

• 分钟线数据支持
• 实时行情接入
• 更多数据源 (东方财富、同花顺)
• 板块/行业分析
• 资金流向分析

---

文档版本: 1.0 最后更新: 2025-11-11 维护者: Claude Code Assistant 状态: 🚧 实施中

---

附录: 文件清单

backend/
├── database/
│   └── schema.sql              # PostgreSQL表结构定义
├── app/
│   ├── database_pg.py          # SQLAlchemy ORM模型
│   └── crawler/
│       └── tushare_crawler.py  # TuShare数据爬虫
docker-compose.yml              # Docker服务编排
DATA_PIPELINE_IMPLEMENTATION.md # 本文档