group_fqcd_jr/tools/seed_sim_account_demo.py

"""场内模拟交易演示种子（§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())