lzf_0626
f6bfcc26b3
## 1. 160129 本就不该在场内清单里 `160129`（南方金利定开债券C）是 `160128`（金利定开债券 **A** 类）的 C 类份额， 而 C 类份额只在场外销售、**不在交易所挂牌** —— 行情源对它永远返回空，下单只能走 净值降级（`source=eastmoney_nav_fallback`），既掩盖了"该产品并不交易"这一事实， 又让成交价带上折溢价偏差。 选型依据：把南方基金全部 **882 个代码**逐个问过腾讯行情源（该源只对交易所上市证券 返回数据），确认真实上市交易的只有 **109 个**；其中 R1/R2 的场内产品 （159700、160128、511070、511810）**原先已在清单内**，所以替换品只能来自 R3 及以上。 选定 **`515450` 红利低波50ETF南方**：仍是南方基金旗下、成交额约 1.3 亿元流动性充足、 红利低波定位偏稳健，与它替换掉的债券 LOF 定位最接近；客户风险等级已覆盖 R3 （原有 `510300` 即 R3）。 改动：`hq.py` 的 `FUND_TYPE_GROUPS`、`tools/import_hq_test_products.py` 的场内映射， 两处都留了注释防止被加回来；`docs/42` / `docs/43` 的产品表与费率表同步 （净值 1.4027、管理费 0.50%、托管费 0.10%）；`docs/42` 另加了一条决策记录。 库内用**复用 `fin_product.id = 9100005`** 的方式替换，使 `fin_holding` / `fin_sim_order` / `fin_transaction` / `fin_market_price` 的既有引用自动跟随， 不产生孤儿数据。那笔历史成交保留 `quote_source='eastmoney_nav_fallback'` —— 它确实是当时的事实，不该被粉饰。同时清掉了 `fin_market_price` 里那条净值降级行， 全库净值降级行情行数归零。 ## 2. 种子假行情会盖住真实行情（更严重，且会反复出现） `seed_sim_account_demo.py` 原来按"今天"upsert 一条假行情（510300=4.50、 510500=6.20，`source='eastmoney_demo_seed'`），而真实行情来自行情源、日期是 **最近交易日**。下单按 `trade_date DESC` 取行情，这条种子行**永远排在真实行情前面**； 它的 `source_updated_at` 只是 seed 运行时刻，过了 `MAX_QUOTE_AGE`（15 分钟）就让整个 产品变成 `503 行情已过期` —— 而库里明明躺着一条刚同步好的真实行情。周末尤其明显： `trade_date` 落在**非交易日**，价格还是编的。 实测表现：`tools/seed_demo_data.py` 跑完第 4 步刚同步过真实行情，冒烟脚本的下单仍报 `503`，且本该 4.579 的成交价被查到 4.50。 改为：**已有任何行情行就绝不插手**（真实行情优先，种子不参与竞争）； 一条都没有时才补，且 `trade_date` 退到最近交易日。 验证：`tools/e2e_smoke_test.py` → **40/40**（下单成交价 4.579 = 真实行情）； ruff 通过；mypy 250 文件 0 错；unit+contract 1381 passed / 0 failed。
302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""场内模拟交易演示种子（§T 用户自助）。
|
||
|
||
按 D4 决策：1 个客户 + 10 万初始资金 + 2 只基金的初始持仓。
|
||
|
||
执行：
|
||
python -m tools.seed_sim_account_demo [--customer-id N] [--quotes-only]
|
||
|
||
## 已知表结构问题（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:
|
||
"""只在**该产品一行行情都没有**时补一条种子行情，供"不跑行情同步也能下单"兜底。
|
||
|
||
⚠️ 这里**刻意不再按"今天"去 upsert**（原实现如此），因为那是一个真实踩过的坑：
|
||
|
||
seed 写的 `trade_date` 是**当天**，而真实行情来自行情源、日期是**最近交易日**。
|
||
下单取行情时按 `trade_date DESC` 排序，于是这条种子行**永远排在真实行情前面**；
|
||
它的 `source_updated_at` 只是 seed 运行时刻，过了 `MAX_QUOTE_AGE`（15 分钟）就让
|
||
整个产品变成 `503 行情已过期` —— 而库里明明躺着一条刚同步好的真实行情。
|
||
周末尤其明显：`trade_date` 落在**非交易日**，价格还是编的（4.50 对不上真实的 4.579）。
|
||
|
||
所以：**已有任何行情行就绝不插手**（真实行情优先，种子不参与竞争）；
|
||
一条都没有时，`trade_date` 也要退到最近交易日，避免再次盖住后续同步的真实行情。
|
||
"""
|
||
existing = (
|
||
await session.execute(
|
||
select(FundMarketPrice.id)
|
||
.where(FundMarketPrice.product_id == product_id)
|
||
.limit(1)
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
return
|
||
today = datetime.now(UTC).date()
|
||
# 周末退到最近的周五，别把 trade_date 写进非交易日
|
||
trade_date = today if today.weekday() < 5 else today - timedelta(days=today.weekday() - 4)
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
next_id = await _next_id(session, FundMarketPrice)
|
||
stmt = insert(FundMarketPrice).values(
|
||
id=next_id,
|
||
product_id=product_id,
|
||
trade_date=trade_date,
|
||
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,
|
||
created_at=now,
|
||
)
|
||
await session.execute(stmt)
|
||
|
||
|
||
async def _upsert_account(
|
||
session: Session, customer_id: int
|
||
) -> tuple[FundSimAccount, bool]:
|
||
existing = (
|
||
await session.execute(
|
||
select(FundSimAccount.id).where(FundSimAccount.customer_id == customer_id)
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
account = (
|
||
await session.execute(
|
||
select(FundSimAccount).where(FundSimAccount.customer_id == customer_id)
|
||
)
|
||
).scalar_one()
|
||
return account, False
|
||
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)
|
||
account = (
|
||
await session.execute(
|
||
select(FundSimAccount).where(FundSimAccount.customer_id == customer_id)
|
||
)
|
||
).scalar_one()
|
||
return account, True
|
||
|
||
|
||
async def _upsert_holding(
|
||
session: Session, customer_id: int, product_id: int, spec: dict
|
||
) -> Decimal:
|
||
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 Decimal("0")
|
||
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)
|
||
return cost
|
||
|
||
|
||
async def run(customer_id: int, *, quotes_only: bool = False) -> 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)} 个 + 当日行情")
|
||
if quotes_only:
|
||
return
|
||
account, account_created = await _upsert_account(session, customer_id)
|
||
added_cost = Decimal("0")
|
||
for pid, spec in zip(product_ids, DEMO_PRODUCTS, strict=True):
|
||
added_cost += await _upsert_holding(session, customer_id, pid, spec)
|
||
if added_cost:
|
||
starting_cash = INITIAL_BALANCE if account_created else account.cash_balance
|
||
account.cash_balance = (starting_cash - added_cost).quantize(Decimal("0.01"))
|
||
account.available_cash = account.cash_balance
|
||
account.updated_at = datetime.now(UTC).replace(tzinfo=None)
|
||
await session.execute(
|
||
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} 初始余额 CNY {INITIAL_BALANCE}")
|
||
print(f" [完成] 持仓已建立，本次新增持仓成本 CNY {added_cost}")
|
||
print(f" [完成] 当前账户现金 CNY {account.cash_balance}")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="场内模拟交易演示种子")
|
||
parser.add_argument("--customer-id", type=int, default=CUSTOMER_ID)
|
||
parser.add_argument(
|
||
"--quotes-only",
|
||
action="store_true",
|
||
help="仅刷新演示产品与当日行情，不修改账户、持仓或余额",
|
||
)
|
||
args = parser.parse_args()
|
||
asyncio.run(run(args.customer_id, quotes_only=args.quotes_only))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|