ljwx-stock/OFFLINE_BATCH_ARCHITECTURE_PLAN.md

离线批处理架构改造方案

创建时间: 2025-12-25 目标: 将"在线实时计算"升级为"离线产出 + 在线读取"的工程级架构

---

📊 现状分析

当前架构（问题明确）

用户请求 → FastAPI
           ↓
    实时抓取行情（TuShare/yfinance）- 3秒
           ↓
    计算技术指标（MA/RSI/MACD）- 5秒
           ↓
    跑ML模型（XGBoost/LightGBM）- 10秒
           ↓
    调用LLM生成解释（Claude API）- 5秒
           ↓
    返回结果 → 用户（总计20-30秒，首次更慢）

核心问题:

• ❌ 首次请求10-20分钟超时（用户无法接受）
• ❌ 每次请求都计算（浪费资源）
• ❌ 无法追溯历史推荐版本
• ❌ 推荐结果不稳定（市场波动导致频繁变化）

数据源识别

数据源	用途	限流	备注
TuShare	A股行情（主要）	500次/分钟	需要token，已配置
yfinance	全球股票/外汇	不明确，但有429限流	免费
akshare	备用数据源	较宽松	免费

现有API接口

端点: GET /api/recommendations

参数:

• limit: 返回数量（可选）
• fast: 是否使用mock数据（可选）

返回结构:

{
  "timestamp": "2025-12-25T15:30:00Z",
  "recommendations": [
    {
      "symbol": "600030",
      "name": "中信证券",
      "score": 99.91,
      "current_price": 20.5,
      "industry": "证券",
      "pe_ratio": 15.2,
      "pb_ratio": 1.8,
      "predicted_return": 7.49,
      "confidence": 0.95,
      "technical_signals": ["MA_Golden_Cross", "RSI_Strong"]
    }
  ],
  "market_context": {
    "overall": "bullish",
    "strength": 0.65,
    "volatility": 0.35
  },
  "explanation": "今日推荐策略基于..."
}

---

🎯 目标架构（工程可用）

核心设计原则

1. 离线计算，在线读取: API永远不做重计算
2. 版本可追溯: 每次推荐都有版本号和时间戳
3. 优雅降级: batch失败时使用昨日数据
4. 双缓冲发布: 避免半成品数据被读到
5. 秒级响应: API响应<300ms，理想<100ms

新架构流程图

离线批处理（夜间 03:30）
  ├─ 增量拉取行情数据（只拉新增交易日）
  ├─ 计算技术指标（全量30只股票）
  ├─ 运行ML模型打分
  ├─ 调用Claude生成解释（可选）
  ├─ 生成推荐结果（JSON）
  ├─ 写入 Postgres（t_recommendation_snapshot）
  └─ 原子发布到 Redis（stock:recommend:latest）

在线API（24/7）
  ├─ 读取 Redis: stock:recommend:latest
  ├─ 若不存在 → 降级读取 Postgres 最新版本
  ├─ 返回结果 + 元数据（版本、生成时间、数据截止日期）
  └─ 响应时间: <100ms

---

🗄️ 数据库设计

1. 推荐快照表

CREATE TABLE t_recommendation_snapshot (
    id BIGSERIAL PRIMARY KEY,

    -- 版本信息
    version VARCHAR(50) NOT NULL,              -- 格式: YYYYMMDD_HHMM 如 20251225_0330
    model_version VARCHAR(20) NOT NULL,         -- 模型版本 如 v1.2.0

    -- 时间信息
    data_date DATE NOT NULL,                    -- 数据截止日期（最后一个交易日）
    generate_time TIMESTAMP NOT NULL,           -- 生成时间
    publish_time TIMESTAMP,                     -- 发布时间（原子切换时）

    -- 推荐结果
    result_json JSONB NOT NULL,                 -- 完整推荐结果
    market_context JSONB,                       -- 市场环境
    explanation TEXT,                           -- LLM生成的解释

    -- 元数据
    stock_universe TEXT[],                      -- 候选股票池（用于追溯）
    feature_summary JSONB,                      -- 特征摘要（便于调试）

    -- 状态
    status VARCHAR(20) NOT NULL DEFAULT 'draft', -- draft/published/archived
    error_message TEXT,                         -- 失败时的错误信息

    -- 审计
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
    created_by VARCHAR(50) DEFAULT 'batch_job',

    -- 索引
    UNIQUE(version),
    INDEX idx_data_date (data_date DESC),
    INDEX idx_status_publish (status, publish_time DESC)
);

COMMENT ON TABLE t_recommendation_snapshot IS '推荐快照表 - 离线批处理产出的推荐结果';
COMMENT ON COLUMN t_recommendation_snapshot.version IS '版本号，格式YYYYMMDD_HHMM，唯一标识一次推荐';
COMMENT ON COLUMN t_recommendation_snapshot.result_json IS '推荐结果JSON，与API返回格式一致';

2. 批处理日志表

CREATE TABLE t_batch_job_log (
    id BIGSERIAL PRIMARY KEY,

    -- 任务信息
    job_name VARCHAR(100) NOT NULL,             -- 如 generate_recommendations
    job_type VARCHAR(50) NOT NULL,              -- daily/manual/retry
    version VARCHAR(50),                        -- 关联的推荐版本

    -- 时间
    start_time TIMESTAMP NOT NULL,
    end_time TIMESTAMP,
    duration_seconds INTEGER,

    -- 状态
    status VARCHAR(20) NOT NULL,                -- running/success/failed/timeout
    stage VARCHAR(50),                          -- 当前阶段（便于定位问题）

    -- 统计
    stocks_processed INTEGER,
    stocks_failed INTEGER,
    api_calls INTEGER,                          -- 外部API调用次数
    cache_hits INTEGER,

    -- 结果
    result_summary JSONB,
    error_message TEXT,
    stack_trace TEXT,

    -- 审计
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),

    INDEX idx_job_name_time (job_name, start_time DESC),
    INDEX idx_version (version)
);

---

🔑 Redis Key 设计

1. 推荐列表（主数据）

Key: stock:recommend:latest

Value (JSON):

{
  "version": "20251225_0330",
  "model_version": "v1.2.0",
  "data_date": "2025-12-24",
  "generate_time": "2025-12-25T03:30:00Z",
  "publish_time": "2025-12-25T03:45:00Z",

  "recommendations": [
    {
      "symbol": "600030",
      "name": "中信证券",
      "score": 99.91,
      "rank": 1,
      "current_price": 20.5,
      "industry": "证券",
      "pe_ratio": 15.2,
      "pb_ratio": 1.8,
      "predicted_return": 7.49,
      "confidence": 0.95,
      "technical_signals": ["MA_Golden_Cross", "RSI_Strong"],
      "risk_level": "medium"
    }
  ],

  "market_context": {
    "overall": "bullish",
    "strength": 0.65,
    "volatility": 0.35,
    "volume": "high",
    "trend": "uptrend"
  },

  "explanation": "今日推荐策略基于多因子模型分析..."
}

TTL: 48小时（避免永久存储，batch失败时仍可使用昨日数据）

---

2. 元数据（用于版本管理）

Key: stock:recommend:meta

Value (JSON):

{
  "current_version": "20251225_0330",
  "previous_version": "20251224_0330",
  "next_scheduled": "2025-12-26T03:30:00Z",
  "last_success_time": "2025-12-25T03:45:00Z",
  "status": "published",
  "message": "Latest recommendation published successfully"
}

TTL: 不过期（需手动管理）

---

3. 双缓冲Key（避免半成品）

临时Key: stock:recommend:next

流程:

1. batch计算完成 → 写入 stock:recommend:next
2. 数据校验通过 → 原子切换：

   # Redis事务
   pipe = redis.pipeline()
   pipe.rename("stock:recommend:next", "stock:recommend:latest")
   pipe.set("stock:recommend:meta", meta_json)
   pipe.execute()
   

3. 校验失败 → 删除 stock:recommend:next，保留 stock:recommend:latest

---

4. 分布式锁（防并发）

Key: lock:stock:batch:generate_recommendations

Value: {hostname}_{pid}_{timestamp}

TTL: 30分钟（batch最大允许运行时间）

使用:

lock_key = "lock:stock:batch:generate_recommendations"
lock_value = f"{hostname}_{os.getpid()}_{int(time.time())}"

# 尝试获取锁
if redis.set(lock_key, lock_value, nx=True, ex=1800):  # 30分钟
    try:
        # 执行batch任务
        generate_recommendations()
    finally:
        # 释放锁（检查值避免误删）
        if redis.get(lock_key) == lock_value:
            redis.delete(lock_key)
else:
    logger.warning("Another batch job is running, skipping...")

---

5. 批处理状态

Key: stock:batch:status

Value (JSON):

{
  "status": "running",
  "stage": "calculating_indicators",
  "start_time": "2025-12-25T03:30:00Z",
  "progress": {
    "total_stocks": 30,
    "processed": 15,
    "failed": 0
  },
  "last_update": "2025-12-25T03:35:12Z"
}

用途:

• batch运行时供监控查看
• API可返回"推荐生成中"状态
• 支持进度条展示

---

🔄 批处理任务设计

核心流程（Python伪代码）

# backend/batch/generate_recommendations.py

import asyncio
from datetime import datetime, date
from typing import Dict, List

class RecommendationBatchJob:
    """推荐批处理任务"""

    def __init__(self):
        self.version = datetime.now().strftime("%Y%m%d_%H%M")
        self.model_version = "v1.2.0"
        self.data_date = None  # 最后一个交易日

    async def run(self):
        """主入口"""
        # 1. 获取分布式锁
        if not self._acquire_lock():
            logger.warning("Another batch is running")
            return

        try:
            # 2. 更新状态
            self._update_status("running", "initializing")

            # 3. 确定数据截止日期（最后一个交易日）
            self.data_date = self._get_last_trade_date()

            # 4. 增量拉取行情数据
            self._update_status("running", "fetching_market_data")
            await self._fetch_incremental_data()

            # 5. 计算技术指标
            self._update_status("running", "calculating_indicators")
            await self._calculate_indicators()

            # 6. 运行ML模型
            self._update_status("running", "running_ml_model")
            recommendations = await self._run_ml_model()

            # 7. 生成市场环境和解释（可选）
            self._update_status("running", "generating_context")
            market_context = await self._get_market_context()
            explanation = await self._generate_explanation(recommendations)

            # 8. 构建结果
            result = self._build_result(recommendations, market_context, explanation)

            # 9. 数据校验
            self._update_status("running", "validating")
            self._validate_result(result)

            # 10. 写入数据库
            self._update_status("running", "persisting")
            snapshot_id = self._save_to_database(result)

            # 11. 双缓冲发布到Redis
            self._update_status("running", "publishing")
            self._publish_to_redis(result)

            # 12. 更新元数据
            self._update_meta()

            # 13. 完成
            self._update_status("success", "completed")
            logger.info(f"✅ Recommendation batch completed: {self.version}")

            return {
                "status": "success",
                "version": self.version,
                "snapshot_id": snapshot_id
            }

        except Exception as e:
            self._update_status("failed", f"error: {str(e)}")
            logger.error(f"❌ Batch failed: {e}", exc_info=True)
            raise
        finally:
            self._release_lock()

    async def _fetch_incremental_data(self):
        """增量拉取行情数据（只拉新增交易日）"""
        # 查询上次数据日期
        last_data_date = self._get_last_data_date_from_db()

        # 只拉取 (last_data_date, self.data_date] 区间的数据
        symbols = self._get_stock_universe()

        for symbol in symbols:
            # 使用现有的 data_fetcher，但限定日期范围
            df = await asyncio.to_thread(
                data_fetcher.fetch_stock_data_range,
                symbol,
                start_date=last_data_date,
                end_date=self.data_date
            )

            # 保存到数据库
            self._save_ohlcv_data(symbol, df)

    async def _calculate_indicators(self):
        """批量计算技术指标"""
        symbols = self._get_stock_universe()

        for symbol in symbols:
            # 读取历史数据（最近60天够计算所有指标）
            df = self._load_ohlcv_data(symbol, days=60)

            # 计算指标
            df_features = feature_engineer.engineer_features(df)

            # 保存到数据库
            self._save_features(symbol, df_features)

    async def _run_ml_model(self) -> List[Dict]:
        """运行ML模型打分"""
        symbols = self._get_stock_universe()
        scores = {}

        for symbol in symbols:
            # 加载特征
            features = self._load_latest_features(symbol)

            # 预测
            score = ml_trainer.predict_score(features)
            scores[symbol] = score

        # 排序并生成推荐列表
        recommendations = self._rank_and_build(scores)
        return recommendations

    def _publish_to_redis(self, result: Dict):
        """双缓冲发布到Redis"""
        # 1. 先写到临时key
        redis.setex(
            "stock:recommend:next",
            172800,  # 48小时
            json.dumps(result, ensure_ascii=False)
        )

        # 2. 数据校验
        verify_data = json.loads(redis.get("stock:recommend:next"))
        assert len(verify_data["recommendations"]) > 0

        # 3. 原子切换
        pipe = redis.pipeline()
        pipe.rename("stock:recommend:next", "stock:recommend:latest")
        pipe.setex(
            "stock:recommend:meta",
            0,  # 不过期
            json.dumps({
                "current_version": self.version,
                "last_success_time": datetime.now().isoformat(),
                "status": "published"
            })
        )
        pipe.execute()

        logger.info(f"✅ Published to Redis: stock:recommend:latest")

    def _validate_result(self, result: Dict):
        """数据校验"""
        assert result["version"] == self.version
        assert len(result["recommendations"]) > 0
        assert result["data_date"] == self.data_date.isoformat()

        # 检查必填字段
        for rec in result["recommendations"]:
            assert "symbol" in rec
            assert "score" in rec
            assert rec["score"] >= 0 and rec["score"] <= 100

---

调度方式（选型建议）

方案1: Python APScheduler（推荐）

优点:

• 与现有FastAPI集成
• 灵活的cron表达式
• 支持异步任务

实现:

# backend/main.py

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from batch.generate_recommendations import RecommendationBatchJob

scheduler = AsyncIOScheduler()

@scheduler.scheduled_job('cron', hour=3, minute=30, timezone='Asia/Shanghai')
async def daily_recommendation_job():
    logger.info("🕐 Starting daily recommendation batch...")
    job = RecommendationBatchJob()
    try:
        result = await job.run()
        logger.info(f"✅ Batch completed: {result}")
    except Exception as e:
        logger.error(f"❌ Batch failed: {e}")
        # 发送告警（飞书/邮件）
        send_alert(f"推荐批处理失败: {e}")

# 在lifespan中启动
@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler.start()
    logger.info("✅ APScheduler started")
    yield
    scheduler.shutdown()

方案2: Celery Beat

优点:

• 分布式任务队列
• 支持重试、失败处理
• 监控工具完善

适用: 如果你已经有Celery环境

方案3: n8n Workflow（你已有）

优点:

• 可视化编排
• 便于非技术人员维护
• 内置告警

配置:

{
  "nodes": [
    {
      "name": "Cron Trigger",
      "type": "n8n-nodes-base.cron",
      "parameters": {
        "triggerTimes": {
          "item": [
            {
              "hour": 3,
              "minute": 30
            }
          ]
        }
      }
    },
    {
      "name": "HTTP Request",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "url": "http://localhost:8028/api/batch/recommendations",
        "method": "POST",
        "timeout": 1800000
      }
    },
    {
      "name": "Send Alert",
      "type": "n8n-nodes-base.feishu",
      "executeOnlyOnce": true,
      "parameters": {
        "resource": "message",
        "webhook": "your_feishu_webhook"
      }
    }
  ]
}

---

🔌 在线API改造

新API设计

# backend/app/routes/recommendations.py

@router.get("/recommendations")
async def get_recommendations(
    limit: Optional[int] = None,
    version: Optional[str] = None  # 新增：支持获取历史版本
):
    """
    获取推荐结果（从Redis/DB读取，永不计算）

    Args:
        limit: 返回数量
        version: 指定版本号（可选，用于追溯历史）

    Returns:
        {
            "version": "20251225_0330",
            "model_version": "v1.2.0",
            "data_date": "2025-12-24",
            "generate_time": "2025-12-25T03:30:00Z",
            "publish_time": "2025-12-25T03:45:00Z",
            "recommendations": [...],
            "market_context": {...},
            "explanation": "...",
            "meta": {
                "source": "redis",  # redis/database/generating
                "cache_age_seconds": 3600,
                "next_update": "2025-12-26T03:30:00Z"
            }
        }
    """
    try:
        # 1. 优先从Redis读取
        if version is None:  # 获取最新版本
            redis_key = "stock:recommend:latest"
            cached = cache.redis_client.get(redis_key)

            if cached:
                result = json.loads(cached)

                # 添加元数据
                result["meta"] = {
                    "source": "redis",
                    "cache_age_seconds": _calculate_cache_age(result["generate_time"]),
                    "next_update": _get_next_schedule()
                }

                # 应用limit
                if limit:
                    result["recommendations"] = result["recommendations"][:limit]

                logger.info(f"📦 Served from Redis: {result['version']}")
                return sanitize_response(result)

        # 2. Redis未命中，降级到数据库
        if version:
            # 获取指定版本
            snapshot = db.query(RecommendationSnapshot).filter_by(
                version=version,
                status="published"
            ).first()
        else:
            # 获取最新版本
            snapshot = db.query(RecommendationSnapshot).filter_by(
                status="published"
            ).order_by(RecommendationSnapshot.publish_time.desc()).first()

        if snapshot:
            result = snapshot.result_json
            result["meta"] = {
                "source": "database",
                "cache_age_seconds": _calculate_cache_age(snapshot.generate_time),
                "next_update": _get_next_schedule(),
                "warning": "Redis缓存未命中，从数据库读取"
            }

            # 应用limit
            if limit:
                result["recommendations"] = result["recommendations"][:limit]

            logger.warning(f"⚠️  Served from database: {snapshot.version}")
            return sanitize_response(result)

        # 3. 数据库也没有，检查batch状态
        batch_status = cache.redis_client.get("stock:batch:status")
        if batch_status:
            status_data = json.loads(batch_status)
            if status_data["status"] == "running":
                # batch正在运行，返回生成中状态
                return {
                    "status": "generating",
                    "message": "推荐正在生成中，请稍后刷新",
                    "progress": status_data.get("progress", {}),
                    "estimated_complete_time": _estimate_complete_time(status_data),
                    "meta": {
                        "source": "generating"
                    }
                }

        # 4. 完全没有数据，返回错误
        raise HTTPException(
            status_code=503,
            detail="推荐数据暂不可用，请联系管理员"
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error getting recommendations: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail=str(e))

---

管理端点（触发手动batch）

@router.post("/admin/batch/recommendations")
async def trigger_batch(
    background_tasks: BackgroundTasks,
    force: bool = False,
    api_key: str = Header(None, alias="X-API-Key")
):
    """
    手动触发推荐批处理

    Args:
        force: 强制执行（即使有锁）
        api_key: 管理员API密钥

    Returns:
        {
            "status": "accepted",
            "job_id": "manual_20251225_1430",
            "message": "Batch job started in background"
        }
    """
    # 验证API密钥
    if api_key != settings.admin_api_key:
        raise HTTPException(status_code=403, detail="Invalid API key")

    # 检查锁
    lock_key = "lock:stock:batch:generate_recommendations"
    if not force and cache.redis_client.exists(lock_key):
        raise HTTPException(
            status_code=409,
            detail="Another batch is already running"
        )

    # 后台执行
    job = RecommendationBatchJob()
    background_tasks.add_task(job.run)

    return {
        "status": "accepted",
        "job_id": f"manual_{job.version}",
        "message": "Batch job started in background"
    }

---

📊 前端适配

推荐页面改进

<template>
  <div class="recommendations">
    <!-- 版本信息条 -->
    <div class="version-bar" v-if="meta">
      <span>数据截止: {{ meta.data_date }}</span>
      <span>生成时间: {{ formatTime(meta.generate_time) }}</span>
      <span>模型版本: {{ meta.model_version }}</span>
      <span>数据年龄: {{ formatAge(meta.cache_age_seconds) }}</span>

      <!-- 管理员可见 -->
      <button v-if="isAdmin" @click="triggerRefresh">
        🔄 重新生成
      </button>
    </div>

    <!-- 生成中状态 -->
    <div v-if="status === 'generating'" class="generating-status">
      <div class="progress-bar">
        <div class="progress" :style="{ width: progress + '%' }"></div>
      </div>
      <p>推荐正在生成中... ({{ progress }}%)</p>
      <p>预计完成时间: {{ estimatedCompleteTime }}</p>
      <button @click="loadRecommendations">刷新</button>
    </div>

    <!-- 推荐列表 -->
    <div v-else class="recommendation-list">
      <div v-for="rec in recommendations" :key="rec.symbol" class="rec-card">
        <!-- 推荐内容 -->
      </div>
    </div>
  </div>
</template>

<script setup>
import { ref, onMounted } from 'vue'
import { fetchRecommendations, triggerBatch } from '@/services/api'

const recommendations = ref([])
const meta = ref(null)
const status = ref('normal')
const progress = ref(0)

const loadRecommendations = async () => {
  try {
    const data = await fetchRecommendations()

    if (data.status === 'generating') {
      status.value = 'generating'
      progress.value = Math.round(
        (data.progress.processed / data.progress.total_stocks) * 100
      )
      estimatedCompleteTime.value = data.estimated_complete_time

      // 5秒后自动重试
      setTimeout(loadRecommendations, 5000)
    } else {
      status.value = 'normal'
      recommendations.value = data.recommendations
      meta.value = data.meta
    }
  } catch (error) {
    console.error('Failed to load recommendations:', error)
  }
}

const triggerRefresh = async () => {
  if (!confirm('确定要重新生成推荐吗？这可能需要10-15分钟')) return

  try {
    await triggerBatch()
    alert('批处理任务已启动，请5分钟后刷新页面')
  } catch (error) {
    alert('触发失败: ' + error.message)
  }
}

onMounted(() => {
  loadRecommendations()
})
</script>

---

🚀 实施路线图

Phase 1: 数据库和Redis（第1周）

目标: 建立离线存储基础

• 创建 t_recommendation_snapshot 表
• 创建 t_batch_job_log 表
• 实现Redis key结构
• 编写数据访问层（DAO）

验收: 能手动写入和读取推荐快照

---

Phase 2: 批处理核心（第2周）

目标: 实现离线计算逻辑

• 实现 RecommendationBatchJob 类
• 增量数据拉取逻辑
• 批量计算指标
• ML模型批量打分
• 双缓冲Redis发布

验收: 能手动执行batch并在Redis看到结果

---

Phase 3: 调度和监控（第3周）

目标: 自动化和可观测

• 集成APScheduler（或n8n）
• 实现分布式锁
• 批处理日志记录
• 失败告警（飞书/邮件）
• 监控面板（可选）

验收: 每天自动执行，失败时收到告警

---

Phase 4: API改造（第4周）

目标: 在线秒级返回

• 改造 GET /recommendations 为只读模式
• 实现降级逻辑（Redis → DB → 生成中）
• 添加 POST /admin/batch/recommendations 管理端点
• 前端适配（版本信息、生成中状态）
• 性能测试（目标<100ms）

验收: API响应<100ms，支持历史版本查询

---

Phase 5: 优化和上线（第5周）

目标: 生产环境验证

• 灰度发布（10% → 50% → 100%）
• 监控指标（响应时间、缓存命中率）
• 压力测试（1000 QPS）
• 文档完善
• 回滚预案

验收: 生产环境稳定运行7天

---

📈 监控指标

批处理监控

指标	目标	告警阈值
执行成功率	100%	<95%
执行时长	<15分钟	>30分钟
推荐数量	10-30只	<5或>50
API调用次数	<200	>500（限流风险）

在线API监控

指标	目标	告警阈值
P50响应时间	<50ms	>100ms
P99响应时间	<200ms	>500ms
Redis命中率	>99%	<90%
错误率	0%	>1%

---

⚠️ 风险和缓解

风险1: Batch失败导致无推荐

缓解:

• 保留昨日推荐（Redis TTL 48小时）
• 降级到数据库最新版本
• ���即告警 + 自动重试（1小时后）

风险2: 数据源限流

缓解:

• 增量拉取（减少API调用）
• 多数据源fallback（TuShare → yfinance → akshare）
• API调用间隔（1秒/只）

风险3: Redis故障

缓解:

• 降级到PostgreSQL
• Redis Sentinel/Cluster（可选）
• 每次batch同时写DB

---

📚 参考文档

• TuShare API文档: https://tushare.pro/document/2
• Redis双缓冲模式: https://redis.io/docs/manual/patterns/pub-sub/
• APScheduler官方文档: https://apscheduler.readthedocs.io/

---

最后更新: 2025-12-25 负责人: 待定 预计上线: 5周后