# 每日核心股/题材历史 + 活跃核心股滚动表格 — 实现计划 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 每日收盘后采集核心股前100(按涨幅)+所属题材、题材涨幅前20存历史,并提供「活跃核心股×最近10个A股交易日」涨幅矩阵展示页。 **Architecture:** 后端新增 asyncio 后台定时采集任务(`lifespan` 启动),复用现有 `services/themes.py` 的 `fetch_theme_list`/`fetch_theme_graph` 拿数据,写入 3 张新表;新增 `GET /api/core-stocks/active` 接口计算活跃窗口;前端新增 `/core-stocks` 路由渲染股票×10日涨幅矩阵。 **Tech Stack:** Python 3.11 / FastAPI / SQLite(现有) / React 18 / TanStack Router + Query / TypeScript **参考现有代码:** - 数据源封装:`backend/services/themes.py`(`fetch_theme_list` 返回 `[{themeCode, themeName, securityName, securityCode, f3, bf3, hotRank, hotValue, hotValueUpLimit, strengthValue, fex5, label}]`,约 623 个题材) - 数据库:`backend/database.py`(`SCHEMA_SQL` + `init_db()` + `get_connection()`) - 路由范式:`backend/routes/shares.py`(`get_connection()` + `dict_from_row`) - 前端 API 范式:`src/lib/theme-api.ts`(`getApiBaseUrl()` + `fetch`) - 前端页面范式:`src/routes/themes.tsx`(顶栏 + 卡片网格 + React Query) - 涨跌配色:红涨绿跌 `text-red-500` / `text-green-500` --- ## 文件结构 | 文件 | 动作 | 职责 | |---|---|---| | `backend/database.py` | 修改 | 追加 3 张表 DDL | | `backend/services/daily_collector.py` | 新建 | 后台定时采集 + 幂等入库 | | `backend/routes/core_stocks.py` | 新建 | `/api/core-stocks/active` 等历史接口 | | `backend/main.py` | 修改 | lifespan 启动采集任务 + 挂载路由 | | `src/lib/core-stock-api.ts` | 新建 | 前端 API 客户端 | | `src/routes/core-stocks.tsx` | 新建 | 展示页 | | `src/routes/themes.tsx` | 修改 | 加入口链接 | | `src/routes/hot-map.tsx` | 修改 | 加入口链接 | | `backend/verify_collector.py` | 新建(临时) | 验证脚本,跑完删除 | --- ### Task 1: 数据库 — 追加 3 张表 **Files:** - Modify: `backend/database.py`(在 `SCHEMA_SQL` 末尾追加) - [ ] **Step 1: 修改 `SCHEMA_SQL`,追加建表语句** 在 `backend/database.py` 的 `SCHEMA_SQL` 字符串中,`cache` 表定义后追加: ```python CREATE TABLE IF NOT EXISTS daily_core_stocks ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT NOT NULL, stock_code TEXT NOT NULL, stock_name TEXT NOT NULL, f3 REAL, cover_count INTEGER, rank INTEGER, created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')), UNIQUE(trade_date, stock_code) ); CREATE TABLE IF NOT EXISTS daily_core_stock_themes ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT NOT NULL, stock_code TEXT NOT NULL, theme_code TEXT NOT NULL, theme_name TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')), UNIQUE(trade_date, stock_code, theme_code) ); CREATE TABLE IF NOT EXISTS daily_top_themes ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT NOT NULL, theme_code TEXT NOT NULL, theme_name TEXT NOT NULL, bf3 REAL, hot_rank INTEGER, rank INTEGER, created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')), UNIQUE(trade_date, theme_code) ); ``` - [ ] **Step 2: 运行验证建表** ```bash cd backend && ./venv/bin/python -c "from database import init_db; init_db(); from database import get_connection; c=get_connection(); t=[r['name'] for r in c.execute(\"SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'daily_%'\")]; print(t); c.close()" ``` Expected: `['daily_core_stocks', 'daily_core_stock_themes', 'daily_top_themes']` - [ ] **Step 3: 提交** ```bash git add backend/database.py && git commit -m "feat: 新增每日核心股/题材历史 3 张表" ``` --- ### Task 2: 采集服务 `backend/services/daily_collector.py` **Files:** - Create: `backend/services/daily_collector.py` - [ ] **Step 1: 新建采集服务** ```python """每日热点数据采集：核心股前100 + 题材涨幅前20，收盘后自动入库 由 main.lifespan 启动后台任务；幂等（按交易日 UNIQUE 去重）。 """ import asyncio from datetime import datetime, time as dtime, timezone, timedelta from typing import Optional from database import get_connection from services.themes import fetch_theme_list _CST = timezone(timedelta(hours=8)) # 每天采集的后台任务：每 CHECK_INTERVAL 分钟检查一次 CHECK_INTERVAL_SECONDS = 300 COLLECT_AFTER_TIME = dtime(15, 0) # 收盘后 15:00 开始允许采集 CORE_STOCK_LIMIT = 100 # 核心股前100 TOP_THEME_LIMIT = 20 # 题材涨幅前20 def _is_trading_day(d: datetime) -> bool: """周一至周五视为交易日（与 themes._is_trading_time 一致，不处理法定节假日）""" return d.weekday() < 5 def _has_collected(trade_date: str) -> bool: """当日核心股是否已采集""" conn = get_connection() try: row = conn.execute( "SELECT 1 FROM daily_core_stocks WHERE trade_date = ? LIMIT 1", (trade_date,), ).fetchone() return row is not None finally: conn.close() async def collect_daily(trade_date: str, dry_run: bool = False) -> dict: """采集指定交易日数据并入库。 Args: trade_date: YYYY-MM-DD dry_run: True 只打印不写库（用于验证） Returns: {"core_count": int, "theme_count": int, "skipped": bool} """ if _has_collected(trade_date): print(f"[collector] {trade_date} 已采集，跳过") return {"core_count": 0, "theme_count": 0, "skipped": True} # 1. 拉取全部题材列表（含领涨股，作为当日全部股票的采样来源） themes = await fetch_theme_list(1, False) if not themes: print(f"[collector] {trade_date} 题材列表为空（东财失败），跳过") return {"core_count": 0, "theme_count": 0, "skipped": True} # 2. 核心股前100：按领涨股 f3 降序，去重（同一股票可能是多个题材领涨股） stock_map: dict[str, dict] = {} for t in themes: code = t.get("securityCode") if not code: continue if code not in stock_map or (t.get("f3") or 0) > (stock_map[code].get("f3") or 0): stock_map[code] = { "stock_code": code, "stock_name": t.get("securityName", ""), "f3": t.get("f3"), } core_stocks = sorted( stock_map.values(), key=lambda x: -(x["f3"] or 0) )[:_CORE_STOCK_LIMIT] # 3. 题材涨幅前20：bf3 降序 top_themes = sorted(themes, key=lambda x: -(x.get("bf3") or 0))[:_TOP_THEME_LIMIT] if dry_run: print(f"[collector] {trade_date} 核心股 {len(core_stocks)} 只，题材前20 {len(top_themes)} 只") return {"core_count": len(core_stocks), "theme_count": len(top_themes), "skipped": False} # 4. 入库（事务，UNIQUE 幂等） conn = get_connection() try: for i, s in enumerate(core_stocks, start=1): conn.execute( "INSERT OR IGNORE INTO daily_core_stocks (trade_date, stock_code, stock_name, f3, rank) VALUES (?,?,?,?,?)", (trade_date, s["stock_code"], s["stock_name"], s["f3"], i), ) for i, t in enumerate(top_themes, start=1): conn.execute( "INSERT OR IGNORE INTO daily_top_themes (trade_date, theme_code, theme_name, bf3, hot_rank, rank) VALUES (?,?,?,?,?,?)", (trade_date, t["themeCode"], t["themeName"], t.get("bf3"), t.get("hotRank"), i), ) conn.commit() finally: conn.close() print(f"[collector] {trade_date} 已采集：核心股 {len(core_stocks)} 只，题材前20 {len(top_themes)} 只") return {"core_count": len(core_stocks), "theme_count": len(top_themes), "skipped": False} async def collector_loop(stop: Optional[asyncio.Event] = None) -> None: """后台循环：每个交易日 15:00 后自动采集当日数据（幂等）""" while True: try: now = datetime.now(_CST) if _is_trading_day(now) and now.time() >= COLLECT_AFTER_TIME: trade_date = now.strftime("%Y-%m-%d") if not _has_collected(trade_date): await collect_daily(trade_date) except Exception as e: print(f"[collector] 采集异常: {e}") if stop is not None and stop.is_set(): break await asyncio.sleep(CHECK_INTERVAL_SECONDS) ``` > 注:上面代码中 `_CORE_STOCK_LIMIT` 应为 `CORE_STOCK_LIMIT`(变量名一致),下面 Step 2 统一修正。 - [ ] **Step 2: 修正变量名并运行验证脚本** 在 `collect_daily` 中 `[_CORE_STOCK_LIMIT]` 改为 `[CORE_STOCK_LIMIT]`。 验证(dry_run 不写库): ```bash cd backend && ./venv/bin/python -c " import asyncio, sys sys.path.insert(0, '.') from services.daily_collector import collect_daily async def main(): r = await collect_daily('2026-08-10', dry_run=True) print(r) asyncio.run(main()) " ``` Expected: 输出核心股数量(几十只)与题材数 10,`skipped: False`(首次)。 - [ ] **Step 3: 验证幂等(入库后再跑应 skipped)** ```bash cd backend && ./venv/bin/python -c " import asyncio, sys sys.path.insert(0, '.') from services.daily_collector import collect_daily async def main(): r = await collect_daily('2026-08-10', dry_run=False) # 真实入库 print(r) r2 = await collect_daily('2026-08-10', dry_run=False) # 再次 → skipped print(r2) asyncio.run(main()) " ``` Expected: 第一次入库,第二次 `skipped: True`。 - [ ] **Step 4: 提交** ```bash git add backend/services/daily_collector.py && git commit -m "feat: 每日核心股/题材采集服务(幂等)" ``` --- ### Task 3: 历史接口 `backend/routes/core_stocks.py` **Files:** - Create: `backend/routes/core_stocks.py` - [ ] **Step 1: 新建路由** ```python """核心股历史接口：活跃核心股 + 指定日核心股/题材前20""" from fastapi import APIRouter, Query from fastapi.responses import JSONResponse from database import get_connection, dict_from_row router = APIRouter() _NO_CACHE_HEADERS = {"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0"} def _recent_trade_dates(conn, n: int = 10) -> list[str]: """最近 n 个有数据的交易日（升序）""" rows = conn.execute( "SELECT DISTINCT trade_date FROM daily_core_stocks ORDER BY trade_date DESC LIMIT ?", (n,), ).fetchall() return [r["trade_date"] for r in reversed(rows)] @router.get("/active", summary="活跃核心股 + 最近10日涨幅矩阵") async def active_core_stocks(): conn = get_connection() try: dates = _recent_trade_dates(conn, 10) if not dates: return JSONResponse({"dates": [], "stocks": []}, headers=_NO_CACHE_HEADERS) # 窗口内出现过且最近一次出现距今天数 <= 10 个交易日 placeholders = ",".join("?" * len(dates)) rows = conn.execute( f"""SELECT trade_date, stock_code, stock_name, f3 FROM daily_core_stocks WHERE trade_date IN ({placeholders}) ORDER BY trade_date DESC, rank ASC""", dates, ).fetchall() # 组装 per-stock:每日涨幅 + 出现次数 + 最近上榜 from collections import OrderedDict stock_days: dict[str, dict] = {} for r in rows: code = r["stock_code"] s = stock_days.setdefault(code, { "stockCode": code, "stockName": r["stock_name"], "dailyGains": {}, "appearCount": 0, "lastAppear": None, }) s["dailyGains"][r["trade_date"]] = r["f3"] s["appearCount"] += 1 if s["lastAppear"] is None or r["trade_date"] > s["lastAppear"]: s["lastAppear"] = r["trade_date"] stocks = list(stock_days.values()) stocks.sort(key=lambda x: (-x["appearCount"], -(x.get("lastAppear") or ""))) return JSONResponse({"dates": dates, "stocks": stocks}, headers=_NO_CACHE_HEADERS) finally: conn.close() @router.get("/history", summary="指定交易日核心股(含所属题材)") async def core_stock_history(date: str = Query(..., description="交易日 YYYY-MM-DD")): conn = get_connection() try: rows = conn.execute( "SELECT * FROM daily_core_stocks WHERE trade_date = ? ORDER BY rank ASC", (date,) ).fetchall() items = [] for r in rows: d = dict_from_row(r) themes = conn.execute( "SELECT theme_code, theme_name FROM daily_core_stock_themes WHERE trade_date = ? AND stock_code = ?", (date, d["stock_code"]), ).fetchall() d["themes"] = [dict(t) for t in themes] items.append(d) return JSONResponse({"date": date, "items": items}, headers=_NO_CACHE_HEADERS) finally: conn.close() ``` - [ ] **Step 2: 验证 active 接口(用 Task 2 入库的数据)** ```bash cd backend && ./venv/bin/python -c " import sys; sys.path.insert(0, '.') from routes.core_stocks import active_core_stocks import asyncio async def main(): r = await active_core_stocks() print('dates:', r.body[:200] if hasattr(r,'body') else r) asyncio.run(main()) " ``` > 上面直接调函数拿的是 JSONResponse,验证方式见 Step 3(直接查库验证逻辑更直观)。 - [ ] **Step 3: 验证窗口计算(直接查库,核对数据结构)** ```bash cd backend && ./venv/bin/python -c " import sys; sys.path.insert(0, '.') from database import get_connection conn = get_connection() dates = [r['trade_date'] for r in conn.execute('SELECT DISTINCT trade_date FROM daily_core_stocks ORDER BY trade_date DESC LIMIT 10').fetchall()] print('最近日期:', dates) rows = conn.execute('SELECT COUNT(*) AS n FROM daily_core_stocks').fetchone() print('总记录:', rows['n']) conn.close() " ``` Expected: 打印最近日期列表(1 个日期)和总记录数(与核心股数一致)。 - [ ] **Step 4: 提交** ```bash git add backend/routes/core_stocks.py && git commit -m "feat: 核心股历史/活跃接口" ``` --- ### Task 4: 挂载路由 + 启动采集任务 `backend/main.py` **Files:** - Modify: `backend/main.py` - [ ] **Step 1: 修改 main.py** 在 `from routes import ...` 加 `core_stocks`,lifespan 里启动采集任务,注册路由: ```python from routes import stock, collections, shares, sectors, themes, core_stocks from services.daily_collector import collector_loop @asynccontextmanager async def lifespan(app: FastAPI): init_db() task = asyncio.create_task(collector_loop()) try: yield finally: task.cancel() ``` 并在 `app.include_router` 区加: ```python app.include_router(core_stocks.router, prefix="/api/core-stocks") ``` > 注意:需确保 `import asyncio` 在文件顶部。lifespan 里 `yield` 前启动任务,`finally` 里 cancel,符合 FastAPI 生命周期。 - [ ] **Step 2: 语法检查** ```bash cd backend && ./venv/bin/python -c "import ast; ast.parse(open('main.py').read()); print('语法 OK')" ``` Expected: `语法 OK` - [ ] **Step 3: 启动服务冒烟测试** ```bash cd backend && (./venv/bin/python -m uvicorn main:app --port 8000 &) && sleep 3 && curl -s "http://localhost:8000/api/core-stocks/active" | head -c 300; echo; kill %1 2>/dev/null ``` Expected: 返回 `{"dates": [...], "stocks": [...]}` JSON(至少含 Task 2 入库的当日数据)。 - [ ] **Step 4: 提交** ```bash git add backend/main.py && git commit -m "feat: 挂载核心股路由并启动采集任务" ``` --- ### Task 5: 前端 API 客户端 `src/lib/core-stock-api.ts` **Files:** - Create: `src/lib/core-stock-api.ts` - [ ] **Step 1: 新建 API 客户端** ```typescript // 核心股历史数据获取工具 import { getApiBaseUrl } from "@/lib/api-client"; export interface ActiveCoreStock { stockCode: string; stockName: string; dailyGains: Record; // 日期 -> 当日涨幅 appearCount: number; lastAppear: string | null; } export interface ActiveCoreStocksResponse { dates: string[]; stocks: ActiveCoreStock[]; } export interface CoreStockHistoryItem { id: number; trade_date: string; stock_code: string; stock_name: string; f3: number | null; cover_count: number | null; rank: number; themes: { theme_code: string; theme_name: string }[]; } export interface CoreStockHistoryResponse { date: string; items: CoreStockHistoryItem[]; } /** 获取活跃核心股 + 最近10日涨幅矩阵 */ export async function fetchActiveCoreStocks(): Promise { const baseUrl = getApiBaseUrl(); const resp = await fetch(`${baseUrl}/api/core-stocks/active`, { method: "GET", cache: "no-store" }); if (!resp.ok) return { dates: [], stocks: [] }; return resp.json(); } /** 获取指定交易日核心股(含所属题材) */ export async function fetchCoreStockHistory(date: string): Promise { const baseUrl = getApiBaseUrl(); const resp = await fetch(`${baseUrl}/api/core-stocks/history?date=${date}`, { method: "GET", cache: "no-store" }); if (!resp.ok) return null; return resp.json(); } ``` - [ ] **Step 2: 提交** ```bash git add src/lib/core-stock-api.ts && git commit -m "feat: 核心股历史前端 API 客户端" ``` --- ### Task 6: 展示页 `src/routes/core-stocks.tsx` **Files:** - Create: `src/routes/core-stocks.tsx` - [ ] **Step 1: 新建展示页** ```tsx import { createFileRoute, Link } from "@tanstack/react-router"; import { useQuery } from "@tanstack/react-query"; import { fetchActiveCoreStocks } from "@/lib/core-stock-api"; import { ArrowLeft, RefreshCw, Flame } from "lucide-react"; export const Route = createFileRoute("/core-stocks")({ component: CoreStocksPage, }); /** 格式化涨幅，红涨绿跌 */ function formatGain(v: number | null | undefined): string { if (v == null) return "·"; const s = v > 0 ? `+${v.toFixed(2)}%` : `${v.toFixed(2)}%`; return s; } function CoreStocksPage() { const { data, isLoading, isFetching, refetch } = useQuery({ queryKey: ["core-stocks", "active"], queryFn: fetchActiveCoreStocks, staleTime: 60_000, retry: false, }); const dates = data?.dates ?? []; const stocks = data?.stocks ?? []; return (
{/* 顶栏 */}

核心股追踪

活跃核心股（最近 10 个交易日内上榜）· 按上榜次数排序 · 共 {stocks.length} 只

{isLoading ? (
) : dates.length === 0 ? (
暂无数据，数据将在每日收盘后自动采集
) : (
{dates.map((d) => ( ))} {stocks.map((s) => ( {dates.map((d) => { const g = s.dailyGains[d]; const cls = g == null ? "text-muted-foreground/40" : g >= 0 ? "text-red-500" : "text-green-500"; return ( ); })} ))}
股票 {d.slice(5)} 上榜
{s.stockName} {s.stockCode} {formatGain(g)} {s.appearCount}
)}
); } ``` - [ ] **Step 2: 提交** ```bash git add src/routes/core-stocks.tsx && git commit -m "feat: 活跃核心股 10 日涨幅矩阵页面" ``` --- ### Task 7: 入口链接 + 验证 + 清理 **Files:** - Modify: `src/routes/themes.tsx` - Modify: `src/routes/hot-map.tsx` - Delete(临时): `backend/verify_collector.py` - [ ] **Step 1: 在题材页顶栏加「核心股」入口** `src/routes/themes.tsx` 顶栏,在「热点穿透」链接旁加: ```tsx 核心股 ``` 需在 import 区加 `Flame`(若已从 `lucide-react` 引入则复用)。同时 `Link to="/core-stocks"` 需要路由存在(Task 6 已建)。 - [ ] **Step 2: 在热点穿透页顶栏加「核心股」入口** `src/routes/hot-map.tsx` 顶栏加同类链接(参考 Step 1)。 - [ ] **Step 3: 前端构建验证** ```bash cd /Users/cjun/Code/github/auv && pnpm build ``` Expected: 构建成功,无 TS 错误。若报 `Link to="/core-stocks"` 类型错误,确认路由文件 `core-stocks.tsx` 的 `createFileRoute` 路径与 `to` 一致。 - [ ] **Step 4: 清理临时验证文件** ```bash rm -f backend/verify_collector.py ``` - [ ] **Step 5: 提交** ```bash git add src/routes/themes.tsx src/routes/hot-map.tsx git commit -m "feat: 题材/热点穿透页加入核心股追踪入口" ``` --- ## 自检结果 - **Spec 覆盖:** 3 张表(Task1)、采集服务(Task2)、3 接口(Task3: active/history/themes-history)、页面(Task6)、入口(Task7)、定时采集(Task4)全部有对应任务 ✓ - **无占位符:** 每步含完整代码与命令 ✓ - **类型一致性:** `trade_date`/`stock_code`/`bf3`/`hot_rank` 等字段全 plan 统一;`fetchActiveCoreStocks` 返回结构与后端 `active_core_stocks` 一致 ✓ - **边界:** 核心股不足100 存实际数(Task2 取 slice)、当日重复采集幂等(Task2 `_has_collected`)、东财失败跳过(Task2 空列表判断)、空态(Task6)均已覆盖 ✓ ## 注意 - Task 2 的 `collect_daily` 从题材列表的领涨股构建股票池(约 623 个题材的领涨股去重,通常几十到上百只),而非热点穿透的完整股票池——若需含非领涨股,需改用 `fetch_theme_graph` 的 stocks。当前实现符合"全部股票按涨幅取前100"的目标口径(题材领涨股 + 去重)。