张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
193 lines
7.6 KiB
Python
193 lines
7.6 KiB
Python
"""把真实行情快照（净值 + 费率）回填到 `fin_product`（默认 dry-run，只有 `--apply` 才写库）。
|
||
|
||
## 为什么需要它
|
||
|
||
`fin_product` 里两个费率字段在本机 **20 只里 19 只为空**，客户问"管理费多少"只能转人工；
|
||
`current_nav` 则有几只被人为改成演示值。这些都能从东方财富取到
|
||
（`EastmoneyFundAdapter.fetch_history` / `fetch_fees`），本脚本负责落到库里。
|
||
|
||
## 口径
|
||
|
||
- **净值取 `DWJZ`（单位净值）**，不是场内交易价。这条口径由既有数据确定：
|
||
库里 19 只的 `current_nav` 与 `fetch_history` 的 `DWJZ` 逐只一致。
|
||
货币 ETF 是唯一容易搞混的 —— 它 `QUOTE_API` 的 `f2` 约 100 元（场内价），
|
||
而 `DWJZ` 是 0.2332 这种量级，**两者不是一回事，不要混用**。
|
||
- 费率取 f10 基金概况页（管理费/托管费），单位是「%/年」。
|
||
|
||
## 安全设计
|
||
|
||
1. **默认 dry-run**：只打印将执行的 UPDATE，不写库。真写要显式 `--apply`。
|
||
2. **默认只补空值**：`NULL` 才写，**已有值一律不动** —— 已有值可能是人为设定。
|
||
净值字段**从来不为空**，所以刷新净值必须显式给 `--overwrite`。
|
||
3. **`--codes` 限定范围**：只回填指定产品，避免"顺手"把全部快照刷成最新
|
||
（全量刷新是 `ProductHistorySyncService` / `MarketQuoteSyncService` 的职责）。
|
||
4. 写入走单个事务，任一失败整体回滚。
|
||
|
||
## 用法
|
||
|
||
python tools/backfill_product_snapshot.py # 看将改什么
|
||
python tools/backfill_product_snapshot.py --apply # 只补空值（费率）
|
||
python tools/backfill_product_snapshot.py --apply --overwrite --codes 510300
|
||
# 连净值和已有费率一起改
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
from datetime import UTC, date, datetime, time
|
||
from pathlib import Path
|
||
|
||
from sqlalchemy import text
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
if str(PROJECT_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from app.infrastructure.db import SessionFactory # noqa: E402
|
||
from app.infrastructure.fund_market_adapter import EastmoneyFundAdapter # noqa: E402
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr]
|
||
|
||
#: 与 `MarketQuoteSyncService.sync` 同一批产品。
|
||
PRODUCT_QUERY = """
|
||
SELECT id, product_code, product_name, current_nav, current_nav_at,
|
||
management_fee_rate, custodian_fee_rate
|
||
FROM fin_product
|
||
WHERE fund_manager = '南方基金'
|
||
AND exchange_code IN ('SSE', 'SZSE')
|
||
AND status = '上市'
|
||
ORDER BY product_code
|
||
"""
|
||
|
||
#: 库里已有值 → 数据源字段。**净值排在最前**，便于阅读。
|
||
FEE_FIELDS = (
|
||
("management_fee_rate", "management_fee_rate"),
|
||
("custodian_fee_rate", "custodian_fee_rate"),
|
||
)
|
||
|
||
#: 净值快照写回时的时间部分：与库里既有记录的 `15:00:00` 保持一致（收盘时刻）。
|
||
NAV_AT_TIME = time(15, 0, 0)
|
||
|
||
|
||
def _same(left: object, right: object) -> bool:
|
||
"""比较两个数值文本，容忍 `0.5` 与 `0.5000`、`4.5` 与 `4.500000` 这种表示差异。"""
|
||
if left is None or right is None:
|
||
return False
|
||
try:
|
||
return abs(float(str(left)) - float(str(right))) < 1e-9
|
||
except (TypeError, ValueError):
|
||
return str(left) == str(right)
|
||
|
||
|
||
async def main() -> int:
|
||
parser = argparse.ArgumentParser(description="把真实净值与费率回填到 fin_product")
|
||
parser.add_argument("--apply", action="store_true", help="真正写库（默认只看不改）")
|
||
parser.add_argument(
|
||
"--overwrite", action="store_true",
|
||
help="连已有值也改（默认只补 NULL，不动人为设定过的值）",
|
||
)
|
||
parser.add_argument(
|
||
"--codes", default="",
|
||
help="只处理这些产品代码（逗号分隔）；留空表示全部",
|
||
)
|
||
args = parser.parse_args()
|
||
wanted = {code.strip() for code in args.codes.split(",") if code.strip()}
|
||
|
||
async with SessionFactory() as session:
|
||
rows = (await session.execute(text(PRODUCT_QUERY))).mappings().all()
|
||
if wanted:
|
||
rows = [row for row in rows if str(row["product_code"]) in wanted]
|
||
if not rows:
|
||
print("没有匹配的产品。")
|
||
return 1
|
||
|
||
adapter = EastmoneyFundAdapter()
|
||
codes = [str(row["product_code"]) for row in rows]
|
||
histories = await asyncio.gather(
|
||
*(adapter.fetch_history(code, date.today()) for code in codes),
|
||
return_exceptions=True,
|
||
)
|
||
fees = await adapter.fetch_fees(codes)
|
||
|
||
print(f"产品 {len(rows)} 只；取到净值 {sum(1 for h in histories if isinstance(h, dict) and not h.get('degraded'))} 只"
|
||
f"、费率 {len(fees)} 只\n")
|
||
|
||
updates: list[tuple[int, str, str, object]] = []
|
||
conflicts: list[str] = []
|
||
for row, history in zip(rows, histories, strict=True):
|
||
code = str(row["product_code"])
|
||
live_nav: object = None
|
||
nav_at: datetime | None = None
|
||
if isinstance(history, dict) and not history.get("degraded"):
|
||
live_nav = history.get("nav")
|
||
nav_date = history.get("nav_date")
|
||
if isinstance(nav_date, date):
|
||
nav_at = datetime.combine(nav_date, NAV_AT_TIME)
|
||
if live_nav is not None:
|
||
stored_nav = row["current_nav"]
|
||
if _same(stored_nav, live_nav):
|
||
pass
|
||
elif stored_nav is not None and not args.overwrite:
|
||
conflicts.append(
|
||
f"{code} {row['product_name']} 的 current_nav："
|
||
f"库里 {stored_nav} → 真实 {live_nav}"
|
||
)
|
||
else:
|
||
updates.append((int(row["id"]), code, "current_nav", live_nav))
|
||
if nav_at is not None:
|
||
updates.append((int(row["id"]), code, "current_nav_at", nav_at))
|
||
|
||
live_fee = fees.get(code) or {}
|
||
for column, key in FEE_FIELDS:
|
||
live_value = live_fee.get(key)
|
||
if live_value is None:
|
||
continue
|
||
stored = row[column]
|
||
if _same(stored, live_value):
|
||
continue
|
||
if stored is not None and not args.overwrite:
|
||
conflicts.append(
|
||
f"{code} {row['product_name']} 的 {column}：库里 {stored} → 真实 {live_value}"
|
||
)
|
||
continue
|
||
updates.append((int(row["id"]), code, column, live_value))
|
||
|
||
if conflicts:
|
||
print("与库里已有值冲突（默认不动，需要 --overwrite 才覆盖）：")
|
||
for line in conflicts:
|
||
print(f" * {line}")
|
||
print()
|
||
|
||
if not updates:
|
||
print("没有需要回填的字段。")
|
||
return 0
|
||
|
||
print(f"将执行 {len(updates)} 条 UPDATE：")
|
||
for _, code, column, value in updates:
|
||
print(f" · {code} {column} = {value}")
|
||
|
||
if not args.apply:
|
||
print("\n[dry-run] 未写库。加 --apply 执行。")
|
||
return 0
|
||
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
async with SessionFactory() as session, session.begin():
|
||
for product_id, _, column, value in updates:
|
||
# 列名来自本模块常量，不是外部输入；值一律走参数绑定。
|
||
await session.execute(
|
||
text(
|
||
f"UPDATE fin_product SET {column} = :value, updated_at = :now "
|
||
"WHERE id = :product_id"
|
||
),
|
||
{"value": value, "now": now, "product_id": product_id},
|
||
)
|
||
print(f"\n已写入 {len(updates)} 个字段。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(asyncio.run(main()))
|