wangjianlong_0626
1f62aca6f7
真机实测到的缺陷：`tools/seed_sim_account_demo.py` 的 `_upsert_market_price()` 在"当天已有行"时直接 `return`，于是同一天重跑种子**不刷新 `source_updated_at`**。 行情是时效数据，过期后 `FundQuoteService` 返回 `503 FUND_QUOTE_UNAVAILABLE：产品 510300 行情已过期`， 连带 `T001` 仪表盘与 `T006` 持仓一起不可用（`T010` 权益不查行情，仍 200）。 表现极具误导性：**刚灌完种子能用，过十几分钟就 503** —— 看起来像行情适配器或缓存故障，实际根因在种子脚本。归因过程： `memory_sync` / Redis / Milvus 全部正常，`fin_market_price` 里当天那行 `source_updated_at` 停在首次灌入时刻。 修法：把列值抽成共享 `values` 字典，当天已有行时 `update` 刷新全部行情列 （含 `source_updated_at`），不存在才 `insert`。 **幂等的正确含义是"不产生重复行"（`(product_id, trade_date)` 唯一）， 不是"不更新值"。** 账户/持仓的"已存在则跳过"保持不变 —— 那是业务数据，不该被种子覆盖。 验证（本机，测试客户 9001 `cust_t`）： - `python -X utf8 -m tools.seed_sim_account_demo --customer-id 9001` 正常 - `GET /api/v1/users/me/account/dashboard` → 200（修前 503） - `GET /api/v1/users/me/holdings` → 200 - `GET /api/v1/users/me/entitlements` → 200 门禁：ruff `app tests tools alembic` 全过；`mypy app` 0 错（252 文件）； `audit_schema.py` 90 张业务表无差异；文档编号/端点编号/RBAC 种子一致性全过。
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""场内模拟交易演示种子（§T 用户自助）。
|
||
|
||
按 D4 决策：1 个客户 + 10 万初始资金 + 2 只基金的初始持仓。
|
||
|
||
执行：
|
||
python -m tools.seed_sim_account_demo [--customer-id N]
|
||
|
||
## 已知表结构问题（2026-09-12 实测）
|
||
|
||
底座 `fin_*` 系列表的 `id` 列**没有** `AUTO_INCREMENT` 属性（仅 `fin_knowledge_meta` 有）。
|
||
SQLAlchemy ORM 默认期望 auto-increment，会在 INSERT 时省略 id → MySQL 报
|
||
`Field 'id' doesn't have a default value`。**底层基础规则不允许 DDL 改动**，
|
||
所以本脚本显式查 `MAX(id) + 1` 分配下一个 id，写入时携带 id（不依赖 auto-increment）。
|
||
|
||
实现：使用 SQLAlchemy Core `insert(...).values(id=..., ...)`，而不是 ORM `session.add()`
|
||
——后者对 BIGINT 不会自动注入 id。
|
||
|
||
幂等：再次执行不会重复建记录（先查 product_code / customer_id / customer_id+product_id）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
from datetime import UTC, datetime, timedelta
|
||
from decimal import Decimal
|
||
|
||
from sqlalchemy import func, insert, select, update
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import get_settings
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.model.fund import (
|
||
FundHolding,
|
||
FundMarketPrice,
|
||
FundProduct,
|
||
FundSimAccount,
|
||
)
|
||
|
||
|
||
CUSTOMER_ID = 9001
|
||
INITIAL_BALANCE = Decimal("100000.00")
|
||
DEMO_PRODUCTS = [
|
||
{
|
||
"product_code": "510300",
|
||
"product_name": "沪深300ETF",
|
||
"exchange_code": "SSE",
|
||
"product_category": "ETF",
|
||
"risk_level": "R3",
|
||
"lot_size": Decimal("100"),
|
||
"price_tick": Decimal("0.001"),
|
||
"close_price": Decimal("4.5000"),
|
||
"total_fund_shares": Decimal("10000000000"),
|
||
"initial_quantity": Decimal("1000"),
|
||
},
|
||
{
|
||
"product_code": "510500",
|
||
"product_name": "南方中证500ETF",
|
||
"exchange_code": "SSE",
|
||
"product_category": "ETF",
|
||
"risk_level": "R3",
|
||
"lot_size": Decimal("100"),
|
||
"price_tick": Decimal("0.001"),
|
||
"close_price": Decimal("6.2000"),
|
||
"total_fund_shares": Decimal("8000000000"),
|
||
"initial_quantity": Decimal("800"),
|
||
},
|
||
]
|
||
|
||
|
||
async def _next_id(session: Session, model) -> int:
|
||
"""返回该表下一个可用的 id（不依赖 AUTO_INCREMENT）。"""
|
||
pk_col = model.__table__.primary_key.columns[0]
|
||
result = await session.execute(select(func.coalesce(func.max(pk_col), 0)))
|
||
return int(result.scalar_one()) + 1
|
||
|
||
|
||
async def _upsert_product(session: Session, spec: dict) -> int:
|
||
existing = (
|
||
await session.execute(
|
||
select(FundProduct.id).where(FundProduct.product_code == spec["product_code"])
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
return int(existing)
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
next_id = await _next_id(session, FundProduct)
|
||
stmt = insert(FundProduct).values(
|
||
id=next_id,
|
||
product_code=spec["product_code"],
|
||
product_name=spec["product_name"],
|
||
exchange_code=spec["exchange_code"],
|
||
product_category=spec["product_category"],
|
||
risk_level=spec["risk_level"],
|
||
fund_manager="南方基金",
|
||
currency="CNY",
|
||
lot_size=spec["lot_size"],
|
||
price_tick=spec["price_tick"],
|
||
current_nav=spec["close_price"],
|
||
current_nav_at=now,
|
||
min_amount=Decimal("100.00"),
|
||
open_start_at=now - timedelta(days=365),
|
||
open_end_at=None,
|
||
transaction_fee_rate=None,
|
||
single_investor_max_holding_ratio=Decimal("5.0000"),
|
||
management_fee_rate=Decimal("0.50"),
|
||
custodian_fee_rate=Decimal("0.10"),
|
||
risk_disclosure_required=0,
|
||
second_confirmation_required=0,
|
||
recording_required=0,
|
||
status="上市",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
await session.execute(stmt)
|
||
return next_id
|
||
|
||
|
||
async def _upsert_market_price(session: Session, product_id: int, spec: dict) -> None:
|
||
"""写入/刷新当日行情。
|
||
|
||
⚠️ 当天已有行时**必须刷新**，不能直接 return（2026-09-12 实测到的缺陷）：
|
||
|
||
行情是**时效数据**，`FundQuoteService` 会按 `source_updated_at` 判新鲜度，
|
||
过期即返回 `503 FUND_QUOTE_UNAVAILABLE`，连带 `T001` 仪表盘与 `T006` 持仓一起不可用。
|
||
原先"已存在就跳过"会让同一天重跑**不更新 `source_updated_at`** ——
|
||
表现为"刚灌完种子能用，过十几分钟仪表盘就 503"，而这与种子无关、极难归因。
|
||
|
||
幂等的正确含义是**不产生重复行**（`(product_id, trade_date)` 唯一），
|
||
**不是"不更新值"**。账户/持仓的"已存在则跳过"是另一回事 ——
|
||
那是业务数据，不该被种子覆盖。
|
||
"""
|
||
today = datetime.now(UTC).date()
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
values = {
|
||
"open_price": spec["close_price"],
|
||
"high_price": spec["close_price"] + Decimal("0.05"),
|
||
"low_price": spec["close_price"] - Decimal("0.05"),
|
||
"close_price": spec["close_price"],
|
||
"volume": Decimal("1000000"),
|
||
"turnover_amount": spec["close_price"] * Decimal("1000000"),
|
||
"total_fund_shares": spec["total_fund_shares"],
|
||
"source": "eastmoney_demo_seed",
|
||
"source_updated_at": now,
|
||
}
|
||
existing = (
|
||
await session.execute(
|
||
select(FundMarketPrice.id).where(
|
||
FundMarketPrice.product_id == product_id,
|
||
FundMarketPrice.trade_date == today,
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
await session.execute(
|
||
update(FundMarketPrice).where(FundMarketPrice.id == existing).values(**values)
|
||
)
|
||
return
|
||
next_id = await _next_id(session, FundMarketPrice)
|
||
await session.execute(
|
||
insert(FundMarketPrice).values(
|
||
id=next_id,
|
||
product_id=product_id,
|
||
trade_date=today,
|
||
created_at=now,
|
||
**values,
|
||
)
|
||
)
|
||
|
||
|
||
async def _upsert_account(session: Session, customer_id: int) -> FundSimAccount:
|
||
existing = (
|
||
await session.execute(
|
||
select(FundSimAccount.id).where(FundSimAccount.customer_id == customer_id)
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
# 返回完整对象
|
||
return (
|
||
await session.execute(
|
||
select(FundSimAccount).where(FundSimAccount.customer_id == customer_id)
|
||
)
|
||
).scalar_one()
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
next_id = await _next_id(session, FundSimAccount)
|
||
stmt = insert(FundSimAccount).values(
|
||
id=next_id,
|
||
account_no=f"FSA{customer_id:06d}",
|
||
customer_id=customer_id,
|
||
currency="CNY",
|
||
cash_balance=INITIAL_BALANCE,
|
||
available_cash=INITIAL_BALANCE,
|
||
frozen_cash=Decimal("0"),
|
||
initial_balance=INITIAL_BALANCE,
|
||
status="正常",
|
||
version=0,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
await session.execute(stmt)
|
||
return (
|
||
await session.execute(
|
||
select(FundSimAccount).where(FundSimAccount.customer_id == customer_id)
|
||
)
|
||
).scalar_one()
|
||
|
||
|
||
async def _upsert_holding(
|
||
session: Session, customer_id: int, product_id: int, spec: dict
|
||
) -> None:
|
||
existing = (
|
||
await session.execute(
|
||
select(FundHolding.id).where(
|
||
FundHolding.customer_id == customer_id,
|
||
FundHolding.product_id == product_id,
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
return
|
||
qty = spec["initial_quantity"]
|
||
cost = (qty * spec["close_price"]).quantize(Decimal("0.01"))
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
next_id = await _next_id(session, FundHolding)
|
||
stmt = insert(FundHolding).values(
|
||
id=next_id,
|
||
customer_id=customer_id,
|
||
trade_account=f"FSA{customer_id:06d}",
|
||
product_id=product_id,
|
||
total_quantity=qty,
|
||
# `shares`/`current_value` 在 docs/00 §6.2 标注为"生成列"，但实测 MySQL 表
|
||
# `GENERATION_EXPRESSION=''`——表结构里是普通 NOT NULL 列、无默认值。
|
||
# 写时必须显式给值，按 §6.2 的语义填（shares=total_quantity, current_value=cost）。
|
||
shares=qty,
|
||
available_quantity=qty,
|
||
frozen_quantity=Decimal("0"),
|
||
average_cost=spec["close_price"],
|
||
cost_amount=cost,
|
||
market_value=None,
|
||
current_value=cost,
|
||
profit_loss=None,
|
||
profit_loss_ratio=None,
|
||
status="持有中",
|
||
first_acquired_at=now,
|
||
version=0,
|
||
updated_at=now,
|
||
)
|
||
await session.execute(stmt)
|
||
|
||
|
||
async def run(customer_id: int) -> None:
|
||
s = get_settings()
|
||
print(f"数据库：{s.mysql_dsn.split('@')[-1]}")
|
||
print(f"目标客户 ID = {customer_id}")
|
||
async with SessionFactory() as session:
|
||
async with session.begin():
|
||
product_ids: list[int] = []
|
||
for spec in DEMO_PRODUCTS:
|
||
pid = await _upsert_product(session, spec)
|
||
await _upsert_market_price(session, pid, spec)
|
||
product_ids.append(pid)
|
||
print(f" ✓ 演示产品 {len(product_ids)} 个 + 当日行情")
|
||
account = await _upsert_account(session, customer_id)
|
||
for pid, spec in zip(product_ids, DEMO_PRODUCTS, strict=True):
|
||
await _upsert_holding(session, customer_id, pid, spec)
|
||
total_cost = sum(
|
||
(spec["initial_quantity"] * spec["close_price"]).quantize(Decimal("0.01"))
|
||
for spec in DEMO_PRODUCTS
|
||
)
|
||
from sqlalchemy import update as sa_update
|
||
account.cash_balance = (INITIAL_BALANCE - total_cost).quantize(Decimal("0.01"))
|
||
account.available_cash = account.cash_balance
|
||
account.updated_at = datetime.now(UTC).replace(tzinfo=None)
|
||
await session.execute(
|
||
sa_update(FundSimAccount)
|
||
.where(FundSimAccount.id == account.id)
|
||
.values(
|
||
cash_balance=account.cash_balance,
|
||
available_cash=account.available_cash,
|
||
updated_at=account.updated_at,
|
||
)
|
||
)
|
||
print(f" ✓ 虚拟账户 {account.account_no} 初始余额 ¥{INITIAL_BALANCE}")
|
||
print(f" ✓ 持仓已建立，账户剩余 ¥{account.cash_balance}（已扣持仓成本 ¥{total_cost}）")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="场内模拟交易演示种子")
|
||
parser.add_argument("--customer-id", type=int, default=CUSTOMER_ID)
|
||
args = parser.parse_args()
|
||
asyncio.run(run(args.customer_id))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |