kyugao/market_sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: PG-only 迁移 + 龙虎榜/tick/moneyflow 同步 + stock_code 统一 + mairui 编码修复

Browse Source 
## 1. PG-only 重构
- 删 app/core/db/connection.py + schema.py (MySQL 路径)
- 新 app/core/db/{orm,models,pg_bootstrap}.py — SQLAlchemy 2.x ORM 一键建表
- 16 张业务表全在 market_data schema,原生 TIMESTAMPTZ / JSONB / Float / TEXT
- requirements.txt 删 PyMySQL 路径,加 psycopg2

## 2. 同步任务扩展(3 个新 task)
- **tick_trade** (mairui hsrl/zbjy):当天逐笔交易,21:00 发布
- **moneyflow** (mairui hsstock/history/transaction):个股资金流,21:30 发布
- **longhubang** (akshare):龙虎榜聚合层 + 席位层(2 张新表)
  - 长虎榜放宽 akshare 政策:仅"无替代源 + 烟测通过"场景允许
  - data_eastmoney 私有 API 不需要(akshare 烟测通过)
- 3-timer 设计:
  - 15:30 market-sync.service (8 base tasks via runall_once)
  - 21:05 market-sync-tick.service (tick_trade)
  - 21:35 market-sync-moneyflow.service (moneyflow)
  - 22:00 market-sync-lhb.service (longhubang,新加)
- bin/systemd/ 新增 tick / moneyflow / lhb 各 1 对 service+timer
- bin/market_sync_*_run.sh wrapper 脚本(不做法定节假日过滤,fail-open)

## 3. stock_code 统一为带 SH/SZ/BJ 前缀
- 历史 bug:stocks.code 用 SH600519,但 kline/moneyflow/tick_trade/kline_5min
  /stock_sector_map/industry 6 张表用纯 6 位 600519,跨表 JOIN 全部 0 行
- 新增 to_hermes() 工具:6位 / 9位(mairui `000001.SZ` 格式)→ 统一 SH000001
- 5 个 task 改写:用 to_hermes(code6) 写入 stock_code
- 一次性迁移 6 张表存量 154M 行(CASE WHEN 探测 + 去重 + 加前缀)
- ORM: stock_sector_map.stock_code / industry.code String(6)→String(10)

## 4. Bug 修复
- **share table stock_code 格式**:之前写 6 位不带前缀,与 stocks 不一致
  → 修 task_share_snapshot + 一次性 UPDATE 63,417 行加前缀
- **share_snapshot warning 状态错填 last_error**:
  → 加 mark_sync_warning() 走专用路径,不写 last_failure_at / last_error
- **schedule config lastRun 不同步**:
  → 加 update_job_status_for_dataset(),SyncTask.run() 完成后自动镜像
  → cli/runall 触发的 task 也能更新 schedule config

## 5. mairui UTF-8 编码修复
- 历史 bug:mairui.py:_fetch 用 latin-1 兜底解码,把所有 UTF-8 中文名
  double-encoded 写入 stocks.name(如 `歌华有线` 变成 `æ\xad\x8cå\x8d\x8e...`)
- 加 _decode_response():UTF-8 → GBK → latin-1 兜底
- 一次性修复 stocks.name 5,213 行:
  - 4,370 行 (encode('latin-1').decode('utf-8') 反向解码)
  - 616 行 (含 fullwidth Ａ,宽松 printable 检查)
  - 820 行 (mid-character 截断,重新从 mairui 拉)

## 6. 测试
- tests/test_smoke.py: TASKS 10→11, SYNC_DEFINITIONS 10→11
- tests/test_schema_models.py: 16→18 张表,新增 longhubang_daily/seat
- pytest 11/11 passed

## 验证
- 6 张表 0 残留无前缀行
- stocks JOIN kline_stock / kline_5min / moneyflow / tick_trade / stock_sector_map:88-100% 命中
- 5,213 stocks.name 全部正确 UTF-8 中文
- pytest 11/11 passed
This commit is contained in:
gao
2026-07-01 22:00:58 +08:00
parent 05635b76b9
commit 7a985dddd5
45 changed files with 3037 additions and 958 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tasks/__init__.py
+4
View File
@@ -7,11 +7,13 @@ from app.tasks.task_industry_sector import SyncIndustrySector
from app.tasks.task_kline_5min import SyncKline5Min
from app.tasks.task_kline_daily import SyncKlineDaily
from app.tasks.task_kline_index import SyncKlineIndex
from app.tasks.task_longhubang import SyncLonghubang
from app.tasks.task_market_regime import SyncMarketRegime
from app.tasks.task_moneyflow import SyncMoneyflow
from app.tasks.task_sector_features import SyncSectorFeatures
from app.tasks.task_share_snapshot import SyncShareSnapshot
from app.tasks.task_stocks_basic import SyncStocksBasic
from app.tasks.task_tick_trade import SyncTickTrade
TASKS: dict[str, type] = {
cls.dataset_id: cls
@@ -20,11 +22,13 @@ TASKS: dict[str, type] = {
SyncKlineDaily,
SyncKlineIndex,
SyncKline5Min,
SyncTickTrade,
SyncMoneyflow,
SyncIndustrySector,
SyncSectorFeatures,
SyncShareSnapshot,
SyncMarketRegime,
SyncLonghubang,
)
}
app/tasks/task_industry_sector.py
+19 -20
View File
@@ -6,9 +6,13 @@ from __future__ import annotations
import time
from typing import Any
from sqlalchemy import update
from app.core.db import ops as db_ops
from app.core.db.models import Stock
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.registry import is_source_ready
from app.core.datasource.utils import to_hermes
from app.core.sync.base import SyncTask
from app.core.sync.registry import mark_sync_blocked
from app.core.utils.logging import get_logger
@@ -65,9 +69,9 @@ class SyncIndustrySector(SyncTask):
"source": "baostock_query_stock_industry",
"enabled": 1,
}
stock_sector_rows.append({"stock_code": code6, "sector_key": sector_key})
stock_sector_rows.append({"stock_code": to_hermes(code6), "sector_key": sector_key})
industry_rows.append({
"code": code6,
"code": to_hermes(code6),
"industry_name": name,
"industry_classification": classification,
"update_date": r.get("update_date", ""),
@@ -82,12 +86,9 @@ class SyncIndustrySector(SyncTask):
return {"status": "error", "message": f"写库失败: {e}"}
# 回填 stocks.industry 字段（让 stocks 表也能直接看到）
# 使用批量 UPDATE ... CASE WHEN 方式一次性完成
# 走 ORM update(Stock) + IN (SHxxx, SZxxx, BJxxx, ...)
try:
from app.core.db.connection import get_mysql
from app.core.db.schema import ensure_all_tables
ensure_all_tables(get_mysql())
# 按 industry_name 分组，对每组用一个 IN 查询批量更新
# 按 industry_name 分组，对每组用一个 IN 批量更新
by_industry: dict[str, list[str]] = {}
for r in industry_rows:
name = r["industry_name"]
@@ -95,19 +96,17 @@ class SyncIndustrySector(SyncTask):
if name not in by_industry:
by_industry[name] = []
by_industry[name].append(code)
with get_mysql().cursor() as cur:
for ind_name, codes in by_industry.items():
# 构造 IN 子句：code IN (SHxxx, SZxxx, BJxxx)
placeholders = []
params = [ind_name]
for c in codes:
for prefix in ("SH", "SZ", "BJ"):
placeholders.append("%s")
params.append(f"{prefix}{c}")
in_clause = ", ".join(placeholders)
sql = f"UPDATE stocks SET industry = %s WHERE code IN ({in_clause})"
cur.execute(sql, params)
get_mysql().commit()
for ind_name, codes in by_industry.items():
# 6 位 code × 3 个交易所前缀（SH/SZ/BJ）
code_variants = [
f"{prefix}{c}" for c in codes for prefix in ("SH", "SZ", "BJ")
]
with db_ops.get_session() as s:
s.execute(
update(Stock)
.where(Stock.code.in_(code_variants))
.values(industry=ind_name)
)
except Exception as e:
logger.warning(f"回填 stocks.industry 失败（不影响主任务）: {e}")
app/tasks/task_kline_5min.py
+7 -5
View File
@@ -18,13 +18,13 @@ from __future__ import annotations
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
from typing import Any, Optional
from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.registry import is_source_ready
from app.core.datasource.utils import is_a_share_code, to_code6
from app.core.datasource.utils import is_a_share_code, to_code6, to_hermes
from app.core.sync.base import SyncTask
from app.core.sync.registry import mark_sync_blocked
from app.core.utils.logging import get_logger
@@ -32,7 +32,7 @@ from app.core.utils.logging import get_logger
logger = get_logger("sync.kline_5min")
# mairui 5min K 实测历史深度起点（实测 2023-06-14 才有数据，更早就 0）
MAIRUI_5MIN_FLOOR = datetime(2023, 6, 14)
MAIRUI_5MIN_FLOOR = datetime(2023, 6, 14, tzinfo=timezone.utc)
# 单次请求窗口大小：1 年（mairui 单次最多 ~11640 条 5min）
WINDOW_DAYS = 365
# 增量时往前多取的天数（防交易日历边界漏当天）
@@ -98,7 +98,8 @@ class SyncKline5Min(SyncTask):
logger.info(f"[5min] DB 里已有 {len(snapshots)} 只的 kline_5min 数据")
# ── 2) 内存算计划 ──
end = datetime.now()
# PG TIMESTAMPTZ 是 timezone-aware，所以 end 也必须 tz-aware 否则比较会炸
end = datetime.now(timezone.utc)
plans = self._plan(stock_codes, end, snapshots)
if not plans:
msg = f"5min K线 — 全部 {len(stock_codes)} 只都已最新（无需同步）"
@@ -174,10 +175,11 @@ class SyncKline5Min(SyncTask):
w_end.strftime("%Y-%m-%d"),
)
if df is not None and not df.empty:
hermes = to_hermes(code6)
for _, r in df.iterrows():
bar_time = r["bar_time"]
all_rows.append({
"stock_code": code6,
"stock_code": hermes,
"bar_time": bar_time.strftime("%Y-%m-%d %H:%M:%S")
if hasattr(bar_time, "strftime") else str(bar_time),
"open": float(r.get("open") or 0),
app/tasks/task_kline_daily.py
+7 -3
View File
@@ -26,7 +26,7 @@ import pandas as pd
from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.utils import is_a_share_code, to_code6
from app.core.datasource.utils import is_a_share_code, to_code6, to_hermes
from app.core.sync.base import SyncTask, _effective_sync_end
from app.core.utils.logging import get_logger
@@ -145,11 +145,14 @@ class SyncKlineDaily(SyncTask):
return (code6, None, str(e)[:120])
if df is None or df.empty:
return (code6, None, "no data")
# 写库用 hermes（带 SH/SZ 前缀）—— 2026-07-01 重构：历史上 kline_stock 用
# 6位 code 与 stocks.code(SH600519) 格式不一致，跨表 JOIN 失败。现统一。
hermes = to_hermes(code6)
rows = []
for _, r in df.iterrows():
td = r["trade_date"]
rows.append({
"stock_code": code6,
"stock_code": hermes,
"trade_date": td.strftime("%Y-%m-%d") if hasattr(td, "strftime") else str(td)[:10],
"open": float(r["open"]),
"high": float(r["high"]),
@@ -226,11 +229,12 @@ class SyncKlineDaily(SyncTask):
return {"status": "fail", "error": str(e)}
if df is None or df.empty:
return {"status": "fail", "error": "no data"}
hermes = to_hermes(code6)
rows = []
for _, r in df.iterrows():
td = r["trade_date"]
rows.append({
"stock_code": code6,
"stock_code": hermes,
"trade_date": td.strftime("%Y-%m-%d") if hasattr(td, "strftime") else str(td)[:10],
"open": float(r["open"]),
"high": float(r["high"]),
app/tasks/task_longhubang.py
+245
View File
@@ -0,0 +1,245 @@
"""同步任务：龙虎榜聚合层 + 席位层（akshare 源）。
数据源：
- 聚合层：`ak.stock_lhb_detail_em(start_date, end_date)` → `longhubang_daily` 表
- 席位层：`ak.stock_lhb_stock_detail_em(symbol, date, flag)` → `longhubang_seat` 表
设计要点：
1) 22:00 触发（mairui 21:30 跑 moneyflow；龙虎榜 19:00-21:00 陆续出齐，22:00 兜底）
2) 默认"今天"增量；CLI 支持 --backfill YYYYMMDD 单日；--backfill-range YYYYMMDD YYYYMMDD 日期范围
3) 聚合层一次性拉日期范围；席位层先抓聚合层取出 stock_code 集合再逐只抓（节省调用）
4) 串行执行（每日调用 ~3 次聚合层 + ~200 次席位层；3-5 分钟内完成）
5) akshare 对"未上榜"的股票会抛 `'NoneType' object is not subscriptable`
→ source 层 try/except 兜住，按"空数据"处理
6) 数据是历史回填友好（全量 4 年 ≈ 25 分钟，2026-07-01 烟测通过）
"""
from __future__ import annotations
import time
from datetime import datetime, timedelta
from typing import Any
from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.registry import is_source_ready
from app.core.sync.base import SyncTask
from app.core.sync.registry import mark_sync_blocked
from app.core.utils.logging import get_logger
logger = get_logger("sync.longhubang")
def _parse_yyyymmdd(s: str) -> str:
"""'YYYYMMDD' 或 'YYYY-MM-DD' → 'YYYY-MM-DD'。"""
s = str(s).strip()
if len(s) == 8 and s.isdigit():
return f"{s[:4]}-{s[4:6]}-{s[6:8]}"
return s
def _date_range(start: str, end: str) -> list[str]:
"""返回 start..end（含）期间所有日期字符串列表（YYYY-MM-DD）。"""
s = datetime.strptime(start, "%Y-%m-%d")
e = datetime.strptime(end, "%Y-%m-%d")
if e < s:
s, e = e, s
out = []
cur = s
while cur <= e:
out.append(cur.strftime("%Y-%m-%d"))
cur += timedelta(days=1)
return out
class SyncLonghubang(SyncTask):
"""龙虎榜聚合层 + 席位层同步。
CLI 用法：
python -m app.entrypoints.cli sync longhubang # 今天增量
... sync longhubang --backfill 20240614 # 单日
... sync longhubang --backfill-range 20240101 20240614 # 日期范围
... sync longhubang --start 20240101 --end 20240614 # 等价于 backfill-range
"""
dataset_id = "longhubang"
def _run(
self,
*,
trigger_source: str = "manual",
backfill: str | None = None,
backfill_range: tuple[str, str] | None = None,
start: str | None = None,
end: str | None = None,
**kwargs,
) -> dict[str, Any]:
src = ds_registry.get("datasource_akshare_lhb")
if src is None:
return {"status": "error", "message": "akshare_lhb 数据源未注册"}
ok, reason = is_source_ready(src.key)
if not ok:
mark_sync_blocked(self.dataset_id, message=f"akshare_lhb 未就绪: {reason}")
return {"status": "blocked", "message": f"akshare_lhb 未就绪: {reason}"}
# ── 决定要拉哪几天 ──
if backfill_range:
s, e = _parse_yyyymmdd(backfill_range[0]), _parse_yyyymmdd(backfill_range[1])
elif start and end:
s, e = _parse_yyyymmdd(start), _parse_yyyymmdd(end)
elif backfill:
s = e = _parse_yyyymmdd(backfill)
else:
# 默认"今天"（盘后未收盘时 fallback 到昨天）
from app.core.sync.base import _effective_sync_end
s = e = _effective_sync_end()
dates = _date_range(s, e)
if not dates:
return {"status": "error", "message": f"日期范围无效: {s}..{e}"}
logger.info(f"[longhubang] 计划: {len(dates)} 天 ({s} ~ {e})")
# ── 逐日同步 ──
t0 = time.time()
daily_ok = daily_fail = 0
daily_total = seat_total = 0
for i, d in enumerate(dates, 1):
self._progress(
message=f"longhubang 进度 {i}/{len(dates)} 当前 {d}",
current=i, total=len(dates), current_step=d,
)
try:
d_rows, s_rows = self._sync_one_day(src, d)
daily_ok += 1
daily_total += d_rows
seat_total += s_rows
except Exception as e:
daily_fail += 1
logger.warning(f"[longhubang {d}] 失败: {e}")
elapsed = round(time.time() - t0, 1)
msg = (
f"龙虎榜 {len(dates)}天 daily_ok={daily_ok} daily_fail={daily_fail} "
f"聚合层={daily_total}行 席位层={seat_total}行, {elapsed}s"
)
return {
"status": "ok" if daily_fail == 0 else "warning",
"message": msg,
"ok": daily_ok, "fail": daily_fail,
"daily_rows": daily_total, "seat_rows": seat_total,
"elapsed_sec": elapsed,
}
# ── 单日同步（聚合 + 席位）──
def _sync_one_day(self, src, trade_date: str) -> tuple[int, int]:
"""返回 (聚合层行数, 席位层行数)。"""
# 1) 聚合层（1 调用/天）
df_daily = src.fetch_longhubang_daily(trade_date, trade_date)
daily_rows = 0
stock_codes: list[str] = []
if df_daily is not None and not df_daily.empty:
rows = []
for _, r in df_daily.iterrows():
rows.append({
"stock_code": str(r.get("stock_code") or "").strip(),
"stock_name": str(r.get("stock_name") or "").strip(),
"trade_date": str(r.get("trade_date") or trade_date)[:10],
"rank_idx": self._to_int(r.get("rank_idx")),
"comment": self._to_str(r.get("comment")),
"close": self._to_float(r.get("close")),
"pct_chg": self._to_float(r.get("pct_chg")),
"lhb_net_buy": self._to_float(r.get("lhb_net_buy")),
"lhb_buy_amt": self._to_float(r.get("lhb_buy_amt")),
"lhb_sell_amt": self._to_float(r.get("lhb_sell_amt")),
"lhb_total_amt": self._to_float(r.get("lhb_total_amt")),
"market_total_amt": self._to_float(r.get("market_total_amt")),
"net_buy_ratio": self._to_float(r.get("net_buy_ratio")),
"total_amt_ratio": self._to_float(r.get("total_amt_ratio")),
"turnover_rate": self._to_float(r.get("turnover_rate")),
"float_mv": self._to_float(r.get("float_mv")),
"reason": self._to_str(r.get("reason")),
"post_1d_pct": self._to_float(r.get("post_1d_pct")),
"post_2d_pct": self._to_float(r.get("post_2d_pct")),
"post_5d_pct": self._to_float(r.get("post_5d_pct")),
"post_10d_pct": self._to_float(r.get("post_10d_pct")),
})
if rows:
db_ops.upsert_longhubang_daily(rows)
daily_rows = len(rows)
# 取出 stock_code（去交易所前缀得 6 位，供席位层调用）
for r in rows:
code6 = str(r["stock_code"]).replace("SH", "").replace("SZ", "").replace("BJ", "")
if len(code6) == 6 and code6.isdigit():
stock_codes.append(code6)
logger.info(f"[longhubang {trade_date}] 聚合层 {daily_rows} 行, {len(stock_codes)} 只上榜")
# 2) 席位层（每只票 2 调用：buy + sell）
seat_rows_total = 0
if stock_codes:
seat_rows = []
for code6 in stock_codes:
for direction in ("buy", "sell"):
df_seat = src.fetch_longhubang_seat(code6, trade_date, direction)
if df_seat is None or df_seat.empty:
continue
for _, r in df_seat.iterrows():
# 字段映射（akshare 6 中文列名）
if direction == "buy":
buy_amt = self._to_float(r.get("买入金额"))
sell_amt = self._to_float(r.get("卖出金额"))
else:
buy_amt = self._to_float(r.get("买入金额"))
sell_amt = self._to_float(r.get("卖出金额"))
# 买方向的占比列叫"买入金额-占总成交比例"，卖方向叫"卖出金额-..."
amt_ratio = (
self._to_float(r.get("买入金额-占总成交比例"))
if direction == "buy"
else self._to_float(r.get("卖出金额-占总成交比例"))
)
# amount=主金额：买方向=买入金额，卖方向=卖出金额
amt = buy_amt if direction == "buy" else sell_amt
seat_rows.append({
"stock_code": f"{'SH' if code6.startswith(('5','6','9')) else 'SZ'}{code6}",
"trade_date": trade_date,
"direction": direction,
"rank_idx": self._to_int(r.get("序号")) or 0,
"branch_name": self._to_str(r.get("交易营业部名称")),
"buy_amount": buy_amt,
"sell_amount": sell_amt,
"amount": amt,
"net_amount": self._to_float(r.get("净额")),
"amount_ratio": amt_ratio,
"explanation": self._to_str(r.get("类型")),
})
if seat_rows:
db_ops.upsert_longhubang_seat(seat_rows)
seat_rows_total = len(seat_rows)
logger.info(f"[longhubang {trade_date}] 席位层 {seat_rows_total} 行")
return daily_rows, seat_rows_total
# ── 类型转换 helper ──
@staticmethod
def _to_float(v) -> float | None:
if v is None:
return None
try:
f = float(v)
return f if f == f else None # NaN → None
except (TypeError, ValueError):
return None
@staticmethod
def _to_int(v) -> int | None:
if v is None:
return None
try:
return int(v)
except (TypeError, ValueError):
return None
@staticmethod
def _to_str(v) -> str | None:
if v is None:
return None
s = str(v).strip()
return s if s else None
app/tasks/task_market_regime.py
+10 -11
View File
@@ -8,8 +8,10 @@ from typing import Any
import numpy as np
import pandas as pd
from sqlalchemy import text
from app.core.db import ops as db_ops
from app.core.db.orm import engine as pg_engine
from app.core.sync.base import SyncTask
from app.core.utils.logging import get_logger
@@ -40,22 +42,19 @@ class SyncMarketRegime(SyncTask):
if not codes:
return {"status": "error", "message": "stocks 表为空"}
# 2. 拉所有 K 线（按股票）- 这里用 SQL GROUP BY 一次性算 daily turnover
from app.core.db.connection import get_mysql
sql = """
# 2. 拉所有 K 线（按股票）— 走 SA text() 走 PG
sql = text("""
SELECT
stock_code,
trade_date,
`close`,
"close",
volume
FROM kline_stock
WHERE trade_date BETWEEN %s AND %s
FROM market_data.kline_stock
WHERE trade_date BETWEEN :start AND :end
ORDER BY stock_code, trade_date
"""
with get_mysql().cursor() as cur:
cur.execute(sql, (start, end))
rows = cur.fetchall()
""")
with pg_engine.connect() as conn:
rows = conn.execute(sql, {"start": start, "end": end}).mappings().all()
if not rows:
return {"status": "warning", "message": f"kline_stock 在 {start} ~ {end} 区间无数据"}
app/tasks/task_moneyflow.py
+2 -2
View File
@@ -15,7 +15,7 @@ from typing import Any
from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.registry import is_source_ready
from app.core.datasource.utils import is_a_share_code, to_code6
from app.core.datasource.utils import is_a_share_code, to_code6, to_hermes
from app.core.sync.base import SyncTask
from app.core.sync.registry import mark_sync_blocked
from app.core.utils.logging import get_logger
@@ -102,7 +102,7 @@ class SyncMoneyflow(SyncTask):
return {"status": "fail", "error": "no data"}
rows = [
{
"stock_code": code6,
"stock_code": to_hermes(code6),
"trade_date": str(r["trade_date"]),
"main_net_inflow": float(r.get("main_net_inflow") or 0),
"large_net_inflow": float(r.get("large_net_inflow") or 0),
app/tasks/task_sector_features.py
+6 -7
View File
@@ -5,7 +5,7 @@
- sector_indices ：行业日线（trade_date × sector_name × close × sector_amplitude）
- sector_features_daily ：行业特征（+ sector_ret + ema10/20/200 + score）
算法（参考 dashboard/api/services/sync/sector.py 重写，持久层改为 MySQL）：
算法（参考 dashboard/api/services/sync/sector.py 重写，持久层改为 PG）：
1) 每只股票日涨跌幅 pct_chg = (close/prev_close - 1) * 100
2) 按行业聚合 sector_ret = 该行业所有股票当日 pct_chg 的均值
3) 行业指数 close = 100 × ∏(1 + sector_ret/100) （基点 100，复合收益）
@@ -26,7 +26,6 @@ from sqlalchemy import create_engine
from app.core.config import settings
from app.core.db import ops as db_ops
from app.core.db.connection import get_mysql
from app.core.sync.base import SyncTask
from app.core.utils.logging import get_logger
@@ -56,11 +55,11 @@ class SyncSectorFeatures(SyncTask):
logger.info("[sector] 启动行业聚合特征计算")
# ── 1) 拉 kline_stock（日线）只取需要的列 ──
# 用 SQLAlchemy engine（pymysql 裸连接 + pd.read_sql 偶发只读 1 行的 bug）
engine = create_engine(settings.mysql_url())
# 走 SQLAlchemy engine + PG URL（pd.read_sql 一次性读 1.1kw 行 → pandas DataFrame）
engine = create_engine(settings.pg_sqlalchemy_url())
df_kline = pd.read_sql(
"SELECT stock_code, trade_date, open, high, low, `close`, volume "
"FROM kline_stock ORDER BY stock_code, trade_date",
'SELECT stock_code, trade_date, open, high, low, "close", volume '
'FROM market_data.kline_stock ORDER BY stock_code, trade_date',
engine,
)
if df_kline.empty:
@@ -72,7 +71,7 @@ class SyncSectorFeatures(SyncTask):
# ── 2) 拉 industry（股票→行业映射）──
df_ind = pd.read_sql(
"SELECT code, industry_name FROM industry WHERE industry_name IS NOT NULL",
"SELECT code, industry_name FROM market_data.industry WHERE industry_name IS NOT NULL",
engine,
)
if df_ind.empty:
app/tasks/task_share_snapshot.py
+5 -3
View File
@@ -87,7 +87,10 @@ class SyncShareSnapshot(SyncTask):
if r is None:
fail_cnt += 1
continue
# 写 stocks
# 写 stocks + share 表 —— 两张表 stock_code 都用 hermes（带 SH/SZ 前缀）
# 与 stocks.code 保持一致，方便 JOIN。
# 历史 bug：之前 share 表写的是裸 6 位 c6，导致 share JOIN stocks 失败。
# 一次性 UPDATE（2026-07-01）已把存量 6.3 万行加上前缀。
hermes = self._to_hermes(c6)
db_ops.update_stock_share_snapshot(
code=hermes,
@@ -95,9 +98,8 @@ class SyncShareSnapshot(SyncTask):
float_share=r["float_share"],
trade_date=r.get("trade_date", today),
)
# 也写 share 表
rows_to_share_table.append({
"stock_code": c6,
"stock_code": hermes,
"trade_date": r.get("trade_date", today),
"total_share": r["total_share"],
"float_share": r["float_share"],
app/tasks/task_tick_trade.py
+168
View File
@@ -0,0 +1,168 @@
"""同步任务：当天逐笔交易（mairui 源）。
数据源：mairui `hsrl/zbjy/{code6}/{licence}`
口径：单笔成交 — 时间 / 价 / 量 / 买卖方向
更新：每日 21:00（mairui 当天数据此时才发布；21:00 之前 API 返空或昨日数据）
设计要点：
1) 起步门控：21:00 之前（未传 force=True）→ 早返 warning
避免 0 行空跑（5000 只股票 × 1s 限速 ≈ 17 分钟浪费）
2) 5 worker × 1 RPS/worker = 5 RPS（守住 mairui 上限）
3) 不做"今日已同步"dedup —— mairui "当天"实际是"最近已发布交易日"，
trade_date 可能比 wall-clock 早一天；重跑靠 PK + ON CONFLICT 幂等
4) 单只股票日均 5w-10w tick，CHUNK=2000 upsert 防 driver 撑爆
5) 数据是"当天 only"，无 start/end 窗口；不存增量概念
"""
from __future__ import annotations
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from typing import Any
from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.registry import is_source_ready
from app.core.datasource.utils import is_a_share_code, to_code6, to_hermes
from app.core.sync.base import SyncTask
from app.core.sync.registry import mark_sync_blocked
from app.core.utils.logging import get_logger
logger = get_logger("sync.tick_trade")
# mairui 文档：「更新：每日 21:00」—— 早于此时 mairui 仍返昨日/空数据
MAIRUI_TICK_TRADE_PUBLISH_HOUR = 21
MAIRUI_TICK_TRADE_PUBLISH_MIN = 0
# 21:00 之后再跑（留 5 分钟缓冲，等 mairui 完整入库）
TICK_TRADE_RUN_GATE = (MAIRUI_TICK_TRADE_PUBLISH_HOUR, MAIRUI_TICK_TRADE_PUBLISH_MIN + 5)
def _passes_publish_gate(now: Optional[datetime] = None, *, force: bool = False) -> tuple[bool, str]:
"""判断当前时间是否已过 mairui 21:00 数据发布门控。
force=True 跳过门控（手动 / 测试用）。
"""
if force:
return True, "force=True 跳过门控"
n = now or datetime.now()
gate = n.replace(hour=TICK_TRADE_RUN_GATE[0], minute=TICK_TRADE_RUN_GATE[1], second=0, microsecond=0)
if n < gate:
return False, f"mairui 逐笔数据 {MAIRUI_TICK_TRADE_PUBLISH_HOUR:02d}:{MAIRUI_TICK_TRADE_PUBLISH_MIN:02d} 发布；当前 {n.strftime('%H:%M')} 早于门控，skip"
return True, "ok"
class SyncTickTrade(SyncTask):
dataset_id = "tick_trade"
def _run(
self,
*,
trigger_source: str = "manual",
codes: list[str] | None = None,
max_workers: int = 5,
force: bool = False,
**kwargs,
) -> dict[str, Any]:
mr = ds_registry.get("datasource_mairui")
if mr is None:
return {"status": "error", "message": "mairui 数据源未注册"}
ok, reason = is_source_ready(mr.key)
if not ok:
mark_sync_blocked(self.dataset_id, message=f"mairui 未就绪: {reason}")
return {"status": "blocked", "message": f"mairui 未就绪: {reason}"}
# ── 21:00 门控（force=True 跳过）──
gate_ok, gate_msg = _passes_publish_gate(force=force)
if not gate_ok:
logger.info(f"[tick_trade] 门控未过: {gate_msg}")
return {"status": "warning", "message": gate_msg, "ok": 0, "fail": 0, "rows": 0, "elapsed_sec": 0.0}
# ── 取股票列表 ──
if codes:
stock_codes = [to_code6(c) for c in codes]
else:
stock_codes = [c for c in db_ops.iter_stock_codes(active_only=True) if is_a_share_code(c)]
limit_raw = os.environ.get("MARKET_DATA_STOCK_LIMIT", "").strip()
if limit_raw.isdigit() and int(limit_raw) > 0:
stock_codes = stock_codes[: int(limit_raw)]
if not stock_codes:
return {"status": "error", "message": "无股票代码"}
# 注：不做"今日已同步"dedup。原因：
# - mairui 的"当天"实际是"最近一个已发布的交易日"（21:00 之前是昨日），
# 用 today_str 做去重会失效（即使刚跑完，trade_date 还是昨日）。
# - PK 含 trade_date，重跑是 ON CONFLICT DO UPDATE，安全且幂等。
# - 17 分钟全市场开销可接受；非交易日用 `MARKET_DATA_STOCK_LIMIT=0` 跳。
total = len(stock_codes)
if total == 0:
return {"status": "ok", "message": "无股票需要同步", "ok": 0, "fail": 0, "rows": 0, "elapsed_sec": 0.0}
# ── 预估耗时：5 RPS 上限 ──
est_sec = total / 5.0
logger.info(f"[tick_trade] 计划: {total} 只, 预估 {est_sec:.0f}s ({est_sec/60:.1f}min)")
self._progress(
message=f"计划: {total} 只, 预估 {est_sec/60:.1f}min @5RPS",
current=0, total=total, current_step="planning",
)
# ── 执行 ──
t0 = time.time()
ok_cnt = fail_cnt = 0
rows_total = 0
with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {pool.submit(self._sync_one, mr, c6): c6 for c6 in stock_codes}
for i, future in enumerate(as_completed(futures), 1):
c6 = futures[future]
try:
res = future.result()
if res["status"] == "ok":
ok_cnt += 1
rows_total += res["rows"]
else:
fail_cnt += 1
if res.get("error"):
logger.warning(f"[tick_trade {c6}] {res['error']}")
except Exception as e:
fail_cnt += 1
logger.warning(f"[tick_trade {c6}] {e}")
if i % 50 == 0 or i == total:
self._progress(
message=f"tick_trade 进度 {i}/{total} OK:{ok_cnt} FAIL:{fail_cnt} 行:{rows_total}",
current=i, total=total, current_step=c6,
)
elapsed = round(time.time() - t0, 1)
msg = f"逐笔 {ok_cnt}成 {fail_cnt}败 共{rows_total}行, {elapsed}s"
return {
"status": "ok" if fail_cnt == 0 else "warning",
"message": msg,
"ok": ok_cnt, "fail": fail_cnt, "rows": rows_total, "elapsed_sec": elapsed,
}
def _sync_one(self, mr, code6: str) -> dict:
try:
df = mr.fetch_tick_trade(code6)
except Exception as e:
return {"status": "fail", "error": str(e)}
if df is None or df.empty:
return {"status": "fail", "error": "no data (mairui 21:00 前可能为空)"}
try:
rows = [
{
"stock_code": to_hermes(code6),
"trade_date": str(r["trade_date"]),
"trade_time": r["trade_time"].strftime("%Y-%m-%d %H:%M:%S")
if hasattr(r["trade_time"], "strftime") else str(r["trade_time"]),
"price": float(r.get("price") or 0),
"volume": float(r.get("volume") or 0),
"direction_code": int(r.get("direction_code") or 0),
"direction": str(r.get("direction") or ""),
"amount": float(r.get("amount") or 0),
}
for _, r in df.iterrows()
]
db_ops.upsert_tick_trade(rows)
return {"status": "ok", "rows": len(rows)}
except Exception as e:
return {"status": "fail", "error": f"db write: {e}"}