ljwx-stock/FULL_DATA_PIPELINE_PLAN.md

完整数据流实施计划

目标: 获取所有A股、美股、港股、外汇、黄金的一年历史数据，实现策略引擎，Dashboard实时显示真实推荐

预计时间: 3-4小时 复杂度: 高

---

📋 需求分析

数据范围

资产类型	数量	时间范围	数据源	数据量估算
A股	~5000只	1年 (250交易日)	TuShare	5000 × 250 = 125万条
美股	精选100只	1年	yfinance	100 × 250 = 2.5万条
港股	精选50只	1年	yfinance	50 × 250 = 1.25万条
外汇	10对	1年 (365天)	Alpha Vantage	10 × 365 = 3650条
黄金/商品	5种	1年 (365天)	yfinance	5 × 365 = 1825条

总数据量: ~150万条记录

存储策略

PostgreSQL表结构:

-- 1. 股票基本信息表
CREATE TABLE stocks (
    symbol VARCHAR(20) PRIMARY KEY,
    name VARCHAR(100),
    market VARCHAR(20),  -- 'A_SHARE', 'US', 'HK', 'FOREX', 'COMMODITY'
    industry VARCHAR(50),
    list_date DATE,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 2. 日线行情表
CREATE TABLE daily_quotes (
    id SERIAL PRIMARY KEY,
    symbol VARCHAR(20),
    trade_date DATE,
    open DECIMAL(10,2),
    high DECIMAL(10,2),
    low DECIMAL(10,2),
    close DECIMAL(10,2),
    volume BIGINT,
    amount DECIMAL(20,2),
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(symbol, trade_date)
);
CREATE INDEX idx_daily_quotes_symbol ON daily_quotes(symbol);
CREATE INDEX idx_daily_quotes_date ON daily_quotes(trade_date);

-- 3. 技术指标表
CREATE TABLE technical_indicators (
    id SERIAL PRIMARY KEY,
    symbol VARCHAR(20),
    trade_date DATE,
    ma5 DECIMAL(10,2),
    ma10 DECIMAL(10,2),
    ma20 DECIMAL(10,2),
    rsi DECIMAL(5,2),
    macd DECIMAL(10,4),
    bollinger_upper DECIMAL(10,2),
    bollinger_lower DECIMAL(10,2),
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(symbol, trade_date)
);

-- 4. 推荐结果表
CREATE TABLE recommendations (
    id SERIAL PRIMARY KEY,
    symbol VARCHAR(20),
    recommend_date DATE,
    strategy VARCHAR(50),
    score DECIMAL(5,2),
    predicted_return DECIMAL(5,2),
    confidence DECIMAL(3,2),
    reason TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_recommendations_date ON recommendations(recommend_date);

API限流考虑

数据源	免费版限制	策略
TuShare	500次/分钟	批量获取，10只/批
yfinance	~2000次/小时	延迟1秒/次
Alpha Vantage	5次/分钟，25次/天	仅关键货币对

预计总时间:

• TuShare (5000只A股): ~10分钟
• yfinance (150只美股/港股): ~3分钟
• Alpha Vantage (10对外汇): ~2分钟

总计: ~15分钟全量加载

---

🏗️ 实施步骤

Phase 1: 数据库准备 (15分钟)

1.1 创建数据库Schema

# 文件: backend/database/schema_v2.sql

1.2 创建数据模型

# 文件: backend/app/models/stock.py
# 使用SQLAlchemy ORM

1.3 数据库迁移

cd backend
alembic revision --autogenerate -m "Add historical data tables"
alembic upgrade head

---

Phase 2: 全量数据获取服务 (45分钟)

2.1 A股全量获取器

# 文件: backend/services/historical_data_loader.py

class HistoricalDataLoader:
    async def load_all_a_shares(self):
        """获取所有A股一年历史数据"""
        # 1. 获取股票列表
        stock_list = await self._get_a_share_list()

        # 2. 批量获取历史数据
        for batch in chunked(stock_list, 10):
            data = await self._fetch_batch_history(batch, period='1y')
            await self._save_to_db(data)
            await asyncio.sleep(1)  # API限流

2.2 美股/港股获取器

class GlobalStockLoader:
    async def load_us_stocks(self, symbols: List[str]):
        """获取美股一年历史"""

    async def load_hk_stocks(self, symbols: List[str]):
        """获取港股一年历史"""

2.3 外汇/商品获取器

class ForexCommodityLoader:
    async def load_forex_history(self, pairs: List[str]):
        """获取外汇一年历史"""

    async def load_commodity_history(self, symbols: List[str]):
        """获取商品一年历史"""

---

Phase 3: 策略引擎 (60分钟)

3.1 策略配置

# 文件: backend/config/strategies.yaml

strategies:
  - name: "价值投资"
    filters:
      - pe_ratio: [0, 15]
      - pb_ratio: [0, 2]
      - roe: [0.1, null]
    weights:
      fundamental: 0.5
      technical: 0.3
      ai: 0.2

  - name: "动量追涨"
    filters:
      - ma5_cross_ma20: true
      - rsi: [50, 70]
    weights:
      technical: 0.7
      volume: 0.3

3.2 策略引擎实现

# 文件: backend/app/modules/strategy_engine.py

class StrategyEngine:
    def __init__(self, config_path: str):
        self.strategies = self._load_strategies(config_path)

    async def calculate_recommendations(
        self,
        strategy_name: str,
        top_n: int = 10
    ) -> List[Recommendation]:
        """根据策略计算推荐"""
        # 1. 应用筛选条件
        candidates = await self._apply_filters(strategy_name)

        # 2. 计算综合评分
        scored = await self._calculate_scores(candidates)

        # 3. 排序并返回Top N
        return sorted(scored, key=lambda x: x.score, reverse=True)[:top_n]

---

Phase 4: 实时推荐API (30分钟)

4.1 修改推荐API

# backend/app/routes/recommendations.py

@router.get("/recommendations")
async def get_recommendations(
    strategy: str = "comprehensive",
    limit: int = 10,
    market: str = "all"  # 'a_share', 'us', 'hk', 'all'
):
    """实时推荐API - 使用真实数据"""

    # 1. 从数据库获取今日数据
    today_data = await db.get_today_quotes()

    # 2. 计算技术指标
    indicators = await calculate_indicators(today_data)

    # 3. 应用策略引擎
    engine = StrategyEngine()
    recommendations = await engine.calculate_recommendations(
        strategy_name=strategy,
        top_n=limit
    )

    # 4. 返回推荐结果
    return {
        "timestamp": datetime.now().isoformat(),
        "strategy": strategy,
        "recommendations": recommendations,
        "data_source": "database"  # 标记使用真实数据
    }

---

Phase 5: Dashboard集成 (30分钟)

5.1 前端API调用

// frontend/src/services/api.js

export async function fetchRealtimeRecommendations(options = {}) {
  const {
    strategy = 'comprehensive',
    limit = 10,
    market = 'all'
  } = options

  const response = await fetch(
    `${API_BASE_URL}/recommendations?strategy=${strategy}&limit=${limit}&market=${market}`
  )
  return response.json()
}

5.2 Dashboard自动刷新

<!-- frontend/src/views/DashboardV2.vue -->

<script setup>
import { ref, onMounted } from 'vue'

const recommendations = ref([])
const autoRefresh = ref(true)
const refreshInterval = 60000  // 1分钟

const loadRecommendations = async () => {
  const data = await fetchRealtimeRecommendations({
    strategy: 'comprehensive',
    limit: 20
  })
  recommendations.value = data.recommendations
}

onMounted(() => {
  loadRecommendations()

  // 自动刷新
  if (autoRefresh.value) {
    setInterval(loadRecommendations, refreshInterval)
  }
})
</script>

---

📊 预期结果

数据库存储

-- 查看数据量
SELECT market, COUNT(*) as stock_count FROM stocks GROUP BY market;
-- A_SHARE: 5000
-- US: 100
-- HK: 50
-- FOREX: 10
-- COMMODITY: 5

SELECT COUNT(*) FROM daily_quotes;
-- ~1,500,000 条

SELECT COUNT(*) FROM recommendations WHERE recommend_date = CURRENT_DATE;
-- ~100 条 (每日推荐)

Dashboard显示

今日推荐:

1. 浦发银行 (600000) - 评分: 95.2 - 策略: 价值投资
   当前价: 8.50 | 涨跌: +2.5% | 推荐理由: PE低于行业平均...

2. 工商银行 (601398) - 评分: 94.8 - 策略: 价值投资
   当前价: 5.20 | 涨跌: +1.8% | 推荐理由: 高ROE，稳定分红...

3. 宁德时代 (300750) - 评分: 93.5 - 策略: 成长投资
   当前价: 195.00 | 涨跌: +3.2% | 推荐理由: 技术突破，量价齐升...

数据来源标记: "基于数据库真实数据 | 最后更新: 2分钟前"

---

⚠️ 风险和挑战

1. 数据量挑战

问题: 150万条数据首次加载耗时长

解决方案:

• 分批加载，显示进度
• 后台任务，不阻塞服务启动
• 增量更新，避免重复获取

2. API限流

问题: TuShare/yfinance可能限流

解决方案:

• 智能延迟控制
• 失败重试机制
• 缓存已获取数据

3. 存储空间

问题: PostgreSQL可能占用大量空间

解决方案:

• 数据压缩（只保留必要字段）
• 定期清理旧数据（保留1年）
• 分区表（按月分区）

4. 计算性能

问题: 实时计算5000只股票指标可能慢

解决方案:

• 预计算技术指标（每日凌晨）
• 使用物化视图
• Redis缓存热门推荐

---

🎯 实施优先级

P0 - 核心功能 (必须完成)

1. ✅ 数据库Schema创建
2. ✅ A股全量数据获取 (至少1000只)
3. ✅ 基本策略引擎
4. ✅ 推荐API修改
5. ✅ Dashboard集成

P1 - 重要功能 (建议完成)

1. 美股/港股数据获取
2. 技术指标预计算
3. 自动刷新机制
4. 数据持久化优化

P2 - 增强功能 (可选)

1. 外汇/商品数据
2. 高级策略配置
3. 回测功能
4. 性能监控

---

📅 时间计划

阶段	任务	预计时间	累计时间
Phase 1	数据库准备	15分钟	15分钟
Phase 2	数据获取服务	45分钟	60分钟
Phase 3	策略引擎	60分钟	120分钟
Phase 4	推荐API	30分钟	150分钟
Phase 5	Dashboard	30分钟	180分钟
测试验证	完整流程测试	30分钟	210分钟

总计: ~3.5小时

---

🚀 快速开始

方案A: 完整实施 (推荐)

# 1. 创建数据库Schema
cd backend
python scripts/init_database.py

# 2. 启动数据加载
python scripts/load_historical_data.py --market a_share --period 1y

# 3. 重启服务
docker-compose restart backend

# 4. 验证
curl http://localhost:8001/api/recommendations?limit=10

方案B: 分步实施 (保守)

第一步: 只获取1000只A股 + 核心策略 第二步: 完善技术指标 + 推荐算法 第三步: 扩展到全部股票 + 多市场

---

📞 下一步行动

立即开始:

1. 创建数据库Schema
2. 实现A股全量获取器
3. 基本策略引擎

需要确认:

• 是否需要所有5000只A股？（推荐先1000只）
• 美股/港股选择哪些？（推荐标普100 + 恒生50）
• 策略配置偏好？（价值/成长/动量）

---

准备好开始实施了吗？

我建议先从核心功能开始：

1. 数据库Schema (15分钟)
2. 1000只A股一年数据 (30分钟)
3. 基本策略引擎 (45分钟)
4. Dashboard集成 (30分钟)

总计约2小时可以看到初步效果。

要开始吗？