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表结构**:
```sql
-- 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
```bash
# 文件: backend/database/schema_v2.sql
```

#### 1.2 创建数据模型
```python
# 文件: backend/app/models/stock.py
# 使用SQLAlchemy ORM
```

#### 1.3 数据库迁移
```bash
cd backend
alembic revision --autogenerate -m "Add historical data tables"
alembic upgrade head
```

---

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

#### 2.1 A股全量获取器
```python
# 文件: 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 美股/港股获取器
```python
class GlobalStockLoader:
    async def load_us_stocks(self, symbols: List[str]):
        """获取美股一年历史"""

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

#### 2.3 外汇/商品获取器
```python
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 策略配置
```yaml
# 文件: 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 策略引擎实现
```python
# 文件: 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
```python
# 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调用
```javascript
// 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自动刷新
```vue
<!-- 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>
```

---

## 📊 预期结果

### 数据库存储

```sql
-- 查看数据量
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: 完整实施 (推荐)

```bash
# 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小时可以看到初步效果。**

要开始吗？