"""按需造数据：给指定客户建持仓，并保证该产品有可查的最新净值。

## 为什么需要它

演示和联调经常冒出"随便给我一只产品、两个客户"的需求。手工写 SQL 会连踩一串坑，
每一条都是这个仓库里真实发生过的：

1. `fin_*` 系列的 `id` 列**没有 AUTO_INCREMENT**（只有 `fin_knowledge_meta` 有），
   不显式给 id 就报 `Field 'id' doesn't have a default value`；
2. `fin_holding.shares` / `current_value` 在 `docs/00` §6.2 标注为"生成列"，
   但实测表里是**普通 NOT NULL 列**（`GENERATION_EXPRESSION=''`），不写就插不进去；
3. 只建 `sys_user` 不绑角色 → 能登录，但查持仓 403（`load_context` 取不到角色就丢权限）；
4. 只建用户不设密码 → `password_hash` 是占位符，**根本登不进来**；
5. 净值散在三处（`fin_product.current_nav`、`fin_nav_history` 最新一条、
   `fin_market_price.close_price`），只写一处就会出现"产品页有净值、持仓页市值为空"；
6. 往 `fin_market_price` 写"今天"的编造行情会**盖住真实行情** —— 下单按
   `trade_date DESC` 取，编造行排在真实行前面，过了 15 分钟`MAX_QUOTE_AGE` 就
   一律 `503 行情已过期`（周末尤其明显）。

本脚本把这些一次性做对，并保证上面第 5 条的三处**严格一致**。

## 用法

    python tools/seed_custom_holdings.py                     # 产品 15911，客户 10001 / 10002
    python tools/seed_custom_holdings.py --show              # 只读：打印现状，不动数据
    python tools/seed_custom_holdings.py --product-code 159911 --customers 10001 10002
    python tools/seed_custom_holdings.py --nav 1.2345 --quantity 2000 --days 250
    python tools/seed_custom_holdings.py --customers 10003 --password abc12345
    python tools/seed_custom_holdings.py --skip-onboarding   # 不完成开户测评（客户会 403）

## 会造出什么

| 表 | 内容 |
|---|---|
| `fin_product` | 产品，`current_nav` = 最新净值 |
| `fin_nav_history` | N 个交易日的净值序列（默认 120） |
| `fin_market_price` | 最近交易日一条行情（**已有行情则绝不插手**） |
| `sys_user` | 客户账号（bcrypt 密码） |
| `sys_user_role` | 绑 `customer` 角色 —— 少了它登录成功但查持仓 403 |
| `fin_sim_account` | 模拟账户 `FSA{客户号:06d}` |
| `fin_holding` | 持仓，市值 = 数量 × 最新净值 |
| `fin_risk_assessment` + 画像 | 开户风险测评，**走接口**，一并生成画像快照与标签 |

最后一行是"能查到"的前提：客户没做测评时，**任何非 onboarding 接口一律 403**
「请先完成开户风险测评问卷」。看起来像权限没配好，其实是合规前置没过。
所以脚本默认会用内置的 13 题答案（C4 成长型）走真实接口完成测评，**而不是**往
`fin_risk_assessment` 插一行充数 —— 那样过得了 403 检查，却不会生成画像与标签。

## 幂等

重复执行不会重复建记录，也不会覆盖已有持仓的数量与成本（只把市值刷成最新净值）；
已存在的用户**不会**被静默重置密码（只有显式传 `--password` 才改）；已完成的测评会跳过。
`--show` 不做任何写入。

## ⚠️ 关于 5 位产品代码

系统现有 20 个产品的 `product_code` **全是 6 位**，且下单要求行情落在
`FundQuoteRuntimeConfig.allowed_codes`（六位数字）白名单里。
所以 5 位代码（如 `15911`）**能建、能查持仓、能查净值，但不能下单**。
如果那是笔误，用 `--product-code 159911` 重跑即可（换代码不会污染已建好的 5 位数据）。
"""

from __future__ import annotations

import argparse
import asyncio
import json
import random
import sys
import urllib.error
import urllib.request
import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from pathlib import Path

from sqlalchemy import func, insert, select, text, update
from sqlalchemy.ext.asyncio import AsyncSession

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.model.fund import (  # noqa: E402
    FundHolding,
    FundMarketPrice,
    FundNavHistory,
    FundProduct,
    FundSimAccount,
)
from app.service.auth_service import hash_password  # noqa: E402

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]

#: 默认值直接对着这次的需求：产品 15911、客户 10001 / 10002。
DEFAULT_PRODUCT_CODE = "15911"
DEFAULT_CUSTOMERS = (10001, 10002)
DEFAULT_PASSWORD = "123456"
DEFAULT_NAV = Decimal("1.0000")
DEFAULT_QUANTITY = Decimal("1000")
DEFAULT_DAYS = 120
#: `customer` 角色在 `sys_role` 里的 id（见 `tools/seed_test_rbac.py`）。
CUSTOMER_ROLE_ID = 9001
#: 账户初始资金，与 `seed_sim_account_demo.py` 保持一致。
INITIAL_BALANCE = Decimal("100000.00")
#: 行情来源标记：一眼能看出这行是种子造的，不是行情源同步的。
SEED_SOURCE = "custom_seed"
#: 接口地址，用于完成开户风险测评。
DEFAULT_BASE_URL = "http://127.0.0.1:8000"

#: 13 道开户测评题的答案（选项序号，1-based）。
#:
#: 按 `risk_questionnaire_service._SCORE_RULES` 折算总分 **46**，落在 41~49 → **C4 成长型**。
#: 这个档位不是随手挑的：产品风险等级是 R3，客户等级必须**不低于**产品等级才能通过
#: 适当性校验，C4 有余量又不至于夸张到 C5。
#:
#: ⚠️ 答案是**服务端评分**的，改选项会改等级：改完请重跑一次，
#: 脚本会把实际拿到的等级打印出来（不要凭算的）。
ONBOARDING_ANSWERS: dict[str, int] = {
    "q1": 1,   # 经济和金融行业人员（5 分）
    "q2": 1,   # 硕士及以上（4 分）
    "q3": 4,   # 利息、股息、转让等金融资产收入（5 分）
    "q4": 3,   # 家庭可支配年收入 50-100 万元（2 分）
    "q5": 2,   # 可用于金融投资的比例 25%~50%（3 分）
    "q6": 1,   # 没有尚未清偿的债务（5 分）
    "q7": 2,   # 大部分投资于基金、股票、信托产品等（4 分）
    "q8": 2,   # 风险投资经验 2 至 5 年（3 分）
    "q9": 2,   # 计划投资 3 至 5 年（3 分）
    "q10": 2,  # 重点投资权益类 + 固定收益类（2 分）
    "q11": 2,  # 寻求较高收益、愿承担有限本金亏损（4 分）
    "q12": 2,  # 大部分投资于高收益高波动的 B（3 分）
    "q13": 2,  # 能承受 30%-50% 的最大损失（3 分）
}


def _call_api(
    base_url: str, method: str, path: str, token: str | None = None, body: dict | None = None
) -> tuple[int, dict]:
    """调一次接口（写操作自动带 Idempotency-Key，与前端行为一致）。"""
    data = json.dumps(body).encode() if body is not None else None
    request = urllib.request.Request(base_url.rstrip("/") + path, data=data, method=method)
    request.add_header("Accept", "application/json")
    if data is not None:
        request.add_header("Content-Type", "application/json")
    if token:
        request.add_header("Authorization", f"Bearer {token}")
    if method != "GET":
        request.add_header("Idempotency-Key", uuid.uuid4().hex)
    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            return response.status, json.loads(response.read().decode())
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode()
        try:
            return exc.code, json.loads(raw)
        except Exception:  # noqa: BLE001
            return exc.code, {"_raw": raw[:200]}
    except Exception as exc:  # noqa: BLE001
        return 0, {"error": {"message": f"{type(exc).__name__}: {exc}"}}


def complete_onboarding(
    base_url: str, user_id: int, password: str
) -> tuple[bool, str]:
    """走**接口**完成开户风险测评，返回 (是否成功, 说明)。

    ⚠️ 为什么不能只在 `fin_risk_assessment` 里插一行充数：
    那样虽然能过 `RiskQuestionnaireService.is_required` 的检查（它只查
    `valid_until > now`），但**不会**生成画像快照、画像标签和记忆同步事件 ——
    而投顾分析、适当性检查、Agent 对话全都要读那些。所以这里走真实接口，
    让业务链路自己把关联数据建全。
    """
    status, payload = _call_api(
        base_url, "POST", "/api/v1/auth/tokens",
        body={"username": str(user_id), "password": password},
    )
    token = (payload.get("data") or {}).get("access_token") if status == 200 else None
    if not token:
        return False, (f"登录失败 HTTP {status}："
                       f"{json.dumps(payload, ensure_ascii=False)[:160]}")

    # 先探一下是否已经完成过：未完成的客户访问非 onboarding 接口会拿到
    # 403「请先完成开户风险测评问卷」。用这个区分"已完成"和"别的问题"。
    status, payload = _call_api(base_url, "GET", "/api/v1/users/me/holdings", token)
    if status == 200:
        return True, "测评已完成（跳过，未重复提交）"
    message = str(((payload.get("error") or {}).get("message")) or "")
    if "风险测评" not in message:
        return False, (f"持仓接口异常 HTTP {status}：{message or json.dumps(payload, ensure_ascii=False)[:160]}")

    status, payload = _call_api(
        base_url,
        "POST",
        "/api/v1/onboarding/risk-questionnaire/submissions",
        token,
        {"answers": ONBOARDING_ANSWERS, "declaration_accepted": True},
    )
    if status != 201:
        return False, (f"提交测评失败 HTTP {status}："
                       f"{json.dumps(payload, ensure_ascii=False)[:200]}")

    data = payload.get("data") or {}
    # ⚠️ 提交返回里**故意没有**风险等级（`_SCORE_RULES` 是服务端机密，
    # 注释写明 "never included in customer responses"）。等级由调用方从库里读回。
    state = data.get("status") or ""
    valid_until = str(data.get("valid_until") or "")[:10]
    return True, f"已完成开户风险测评（状态 {state}，有效期至 {valid_until or '?'}）"


def _now() -> datetime:
    return datetime.now(UTC).replace(tzinfo=None)


def _recent_trading_days(count: int) -> list[date]:
    """最近的 `count` 个交易日（只跳周末，不含节假日 —— 够用且可解释）。

    刻意不把净值日期落在周末：`fin_nav_history` 的折线图会因此凭空多出两个点。
    """
    today = datetime.now(UTC).date()
    # 周末退到最近的周五
    day = today if today.weekday() < 5 else today - timedelta(days=today.weekday() - 4)
    days: list[date] = []
    while len(days) < count:
        if day.weekday() < 5:
            days.append(day)
        day -= timedelta(days=1)
    days.reverse()
    return days


def _build_nav_series(latest: Decimal, days: int) -> list[tuple[date, Decimal]]:
    """以 `latest` 收尾、往前倒推 `days` 个交易日的净值序列。

    **确定性**：随机数用固定种子，同一组入参永远得到同一条曲线 ——
    否则每次重跑都改变历史净值，对着同一份数据排查问题就失去了意义。
    """
    rng = random.Random(20260914)
    navs: list[Decimal] = [latest]
    for _ in range(days - 1):
        # 相对前一日 ±1.2% 的波动，倒推出前一天
        change = Decimal(str(round(rng.uniform(-0.012, 0.012), 6)))
        previous = (navs[-1] / (Decimal("1") + change)).quantize(Decimal("0.000001"))
        navs.append(previous)
    navs.reverse()
    return list(zip(_recent_trading_days(days), navs, strict=True))


async def _next_id(session: AsyncSession, model: type) -> int:
    """该表下一个可用 id —— 这些表没有 AUTO_INCREMENT，必须自己分配。"""
    pk = model.__table__.primary_key.columns[0]  # type: ignore[attr-defined]
    result = await session.execute(select(func.coalesce(func.max(pk), 0)))
    return int(result.scalar_one()) + 1


# ------------------------------------------------------------------ 产品与净值

async def ensure_product(
    session: AsyncSession,
    code: str,
    *,
    name: str,
    exchange: str,
    category: str,
    risk_level: str,
    nav: Decimal,
) -> tuple[int, bool]:
    existing = (
        await session.execute(select(FundProduct.id).where(FundProduct.product_code == code))
    ).scalar_one_or_none()
    now = _now()
    if existing is not None:
        # 已存在：只把最新净值刷成目标值，其余字段不动（不覆盖别人维护的产品资料）
        await session.execute(
            update(FundProduct)
            .where(FundProduct.id == int(existing))
            .values(current_nav=nav, current_nav_at=now, updated_at=now)
        )
        return int(existing), False

    product_id = await _next_id(session, FundProduct)
    await session.execute(
        insert(FundProduct).values(
            id=product_id,
            product_code=code,
            product_name=name,
            exchange_code=exchange,
            product_category=category,
            risk_level=risk_level,
            fund_manager="自定义",
            currency="CNY",
            lot_size=Decimal("100"),
            price_tick=Decimal("0.001"),
            current_nav=nav,
            current_nav_at=now,
            min_amount=Decimal("100.00"),
            open_start_at=now - timedelta(days=365),
            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,
        )
    )
    return product_id, True


async def ensure_nav_history(
    session: AsyncSession, product_id: int, latest: Decimal, days: int
) -> int:
    """补齐缺失的净值行，返回本次新增条数。已存在的日期不动。"""
    existing = set(
        (
            await session.execute(
                select(FundNavHistory.nav_date).where(FundNavHistory.product_id == product_id)
            )
        )
        .scalars()
        .all()
    )
    series = _build_nav_series(latest, days)
    added = 0
    for nav_date, nav in series:
        if nav_date in existing:
            continue
        await session.execute(
            insert(FundNavHistory).values(
                id=await _next_id(session, FundNavHistory),
                product_id=product_id,
                nav_date=nav_date,
                nav=nav,
                created_at=_now(),
            )
        )
        added += 1
    return added


async def ensure_market_price(session: AsyncSession, product_id: int, nav: Decimal) -> str:
    """给"一条行情都没有"的产品补一条，供不跑行情同步也能下单。

    ⚠️ **已有任何行情行就绝不插手** —— 真实行情优先，种子不参与竞争。
    这条规则是被坑出来的：种子按"今天"写行情，而真实行情的 `trade_date` 是最近交易日，
    下单按 `trade_date DESC` 取，于是种子行永远排在前面；它的 `source_updated_at` 只是
    种子运行时刻，过了 `MAX_QUOTE_AGE`（15 分钟）整个产品就 `503 行情已过期`，
    而库里明明躺着一条刚同步好的真实行情。
    """
    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 "已有行情，跳过（不覆盖真实行情）"
    now = _now()
    trade_date = _recent_trading_days(1)[0]
    await session.execute(
        insert(FundMarketPrice).values(
            id=await _next_id(session, FundMarketPrice),
            product_id=product_id,
            trade_date=trade_date,
            open_price=nav,
            high_price=nav,
            low_price=nav,
            close_price=nav,
            change_pct=Decimal("0.0000"),
            volume=Decimal("0"),
            turnover_amount=Decimal("0.00"),
            total_fund_shares=Decimal("1000000000.0000"),
            source=SEED_SOURCE,
            source_updated_at=now,
            created_at=now,
        )
    )
    return f"新增 {trade_date} 一条（source={SEED_SOURCE}）"


# ------------------------------------------------------------------ 用户与账户

async def ensure_user(
    session: AsyncSession, user_id: int, *, password: str, password_changed: bool
) -> tuple[str, str]:
    row = (
        await session.execute(
            text("SELECT id, password_hash FROM sys_user WHERE id = :u"), {"u": user_id}
        )
    ).mappings().one_or_none()
    now = _now()

    if row is None:
        await session.execute(
            text(
                "INSERT INTO sys_user (id, user_no, username, password_hash, user_type, "
                "is_professional_investor, professional_investor_status, fund_account_status, "
                "fund_account_opened_at, status, created_at, updated_at) VALUES "
                "(:id, :no, :name, :hash, 'customer', 0, 'none', '已开户', :now, '正常', :now, :now)"
            ),
            {
                "id": user_id,
                "no": str(user_id),
                "name": str(user_id),
                "hash": hash_password(password),
                "now": now,
            },
        )
        user_state = "新建"
    else:
        # 已存在：只在明确要求时改密码，绝不静默重置
        if password_changed:
            await session.execute(
                text("UPDATE sys_user SET password_hash = :h, updated_at = :now WHERE id = :id"),
                {"h": hash_password(password), "now": now, "id": user_id},
            )
            user_state = "已存在，密码按 --password 重设"
        else:
            user_state = "已存在（密码未动）"
        # 补齐可能缺失的客户属性，否则登录后适当性/账户链路会认为"不是客户"
        await session.execute(
            text(
                "UPDATE sys_user SET user_type = 'customer', status = '正常', "
                "fund_account_status = '已开户', updated_at = :now WHERE id = :id"
            ),
            {"now": now, "id": user_id},
        )

    # ⚠️ 没绑角色就查不到持仓：`load_context` 取不到角色 → 权限集为空 → 接口 403。
    bound = (
        await session.execute(
            text("SELECT id FROM sys_user_role WHERE user_id = :u AND role_id = :r"),
            {"u": user_id, "r": CUSTOMER_ROLE_ID},
        )
    ).scalar_one_or_none()
    if bound is None:
        await session.execute(
            text(
                "INSERT INTO sys_user_role (user_id, role_id, assigned_at) "
                "VALUES (:u, :r, :now)"
            ),
            {"u": user_id, "r": CUSTOMER_ROLE_ID, "now": now},
        )
        role_state = f"已绑 customer 角色({CUSTOMER_ROLE_ID})"
    else:
        role_state = "角色已绑定"
    return user_state, role_state


async def ensure_account(session: AsyncSession, customer_id: int) -> tuple[str, Decimal, bool]:
    row = (
        await session.execute(
            select(FundSimAccount.id, FundSimAccount.account_no, FundSimAccount.cash_balance)
            .where(FundSimAccount.customer_id == customer_id)
        )
    ).one_or_none()
    if row is not None:
        return str(row[1]), Decimal(str(row[2])), False

    now = _now()
    account_no = f"FSA{customer_id:06d}"
    await session.execute(
        insert(FundSimAccount).values(
            id=await _next_id(session, FundSimAccount),
            account_no=account_no,
            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,
        )
    )
    return account_no, INITIAL_BALANCE, True


# ------------------------------------------------------------------ 持仓

async def ensure_holding(
    session: AsyncSession,
    customer_id: int,
    trade_account: str,
    product_id: int,
    *,
    quantity: Decimal,
    nav: Decimal,
) -> tuple[str, Decimal, Decimal]:
    """建持仓；已存在则**只把市值刷成最新净值**，不动数量与成本。

    返回 (状态, 数量, 市值)。市值恒等于 `数量 × 最新净值` —— 这是"三处一致"的落点：
    净值改了，持仓市值跟着改，不会出现"产品页 1.2345、持仓页还按 1.0000 算"。
    """
    market_value = (quantity * nav).quantize(Decimal("0.01"))
    row = (
        await session.execute(
            select(FundHolding.id, FundHolding.total_quantity)
            .where(
                FundHolding.customer_id == customer_id,
                FundHolding.product_id == product_id,
            )
        )
    ).one_or_none()

    if row is not None:
        held = Decimal(str(row[1]))
        await session.execute(
            update(FundHolding)
            .where(FundHolding.id == int(row[0]))
            .values(
                market_value=(held * nav).quantize(Decimal("0.01")),
                current_value=(held * nav).quantize(Decimal("0.01")),
                updated_at=_now(),
            )
        )
        return "已存在，仅刷新市值", held, (held * nav).quantize(Decimal("0.01"))

    cost = (quantity * nav).quantize(Decimal("0.01"))
    now = _now()
    await session.execute(
        insert(FundHolding).values(
            id=await _next_id(session, FundHolding),
            customer_id=customer_id,
            trade_account=trade_account,
            product_id=product_id,
            total_quantity=quantity,
            # `shares` / `current_value` 必须显式写：文档说是生成列，实际是普通 NOT NULL 列
            shares=quantity,
            available_quantity=quantity,
            frozen_quantity=Decimal("0"),
            average_cost=nav,
            cost_amount=cost,
            market_value=market_value,
            current_value=market_value,
            profit_loss=Decimal("0.00"),
            profit_loss_ratio=Decimal("0.0000"),
            status="持有中",
            first_acquired_at=now,
            version=0,
            updated_at=now,
        )
    )
    return "新建", quantity, market_value


# ------------------------------------------------------------------ 只读汇总

async def show_summary(session: AsyncSession, code: str, customers: list[int]) -> None:
    product = (
        await session.execute(
            text(
                "SELECT id, product_code, product_name, exchange_code, risk_level, "
                "current_nav, current_nav_at, status FROM fin_product WHERE product_code = :c"
            ),
            {"c": code},
        )
    ).mappings().one_or_none()

    print(f"\n===== 产品 {code} =====")
    if product is None:
        print("  （不存在）")
        return
    print(f"  id={product['id']}  {product['product_name']}  {product['exchange_code']}  "
          f"{product['risk_level']}  {product['status']}")
    print(f"  current_nav = {product['current_nav']}  （更新于 {product['current_nav_at']}）")

    nav_row = (
        await session.execute(
            text(
                "SELECT nav_date, nav FROM fin_nav_history WHERE product_id = :p "
                "ORDER BY nav_date DESC LIMIT 3"
            ),
            {"p": product["id"]},
        )
    ).all()
    total_nav = (
        await session.execute(
            text("SELECT COUNT(*) FROM fin_nav_history WHERE product_id = :p"),
            {"p": product["id"]},
        )
    ).scalar()
    print(f"  净值历史共 {total_nav} 条，最近 3 条："
          + "  ".join(f"{d}={n}" for d, n in nav_row))

    price = (
        await session.execute(
            text(
                "SELECT trade_date, close_price, change_pct, source FROM fin_market_price "
                "WHERE product_id = :p ORDER BY trade_date DESC LIMIT 1"
            ),
            {"p": product["id"]},
        )
    ).mappings().one_or_none()
    print(f"  最新行情：{dict(price) if price else '（无）'}")

    latest_nav = Decimal(str(product["current_nav"])) if product["current_nav"] else None
    consistent = (
        bool(nav_row) and latest_nav is not None and Decimal(str(nav_row[0][1])) == latest_nav
    )
    print(f"  [一致性] current_nav 与净值最新一条"
          f"{'一致' if consistent else '不一致 —— 需要重跑脚本'}")

    print("\n===== 客户持仓 =====")
    for customer_id in customers:
        user = (
            await session.execute(
                text(
                    "SELECT id, user_no, username, user_type, status, fund_account_status "
                    "FROM sys_user WHERE id = :u"
                ),
                {"u": customer_id},
            )
        ).mappings().one_or_none()
        if user is None:
            print(f"\n  客户 {customer_id}：（用户不存在）")
            continue
        roles = (
            await session.execute(
                text(
                    "SELECT r.role_code FROM sys_user_role ur JOIN sys_role r ON r.id = ur.role_id "
                    "WHERE ur.user_id = :u"
                ),
                {"u": customer_id},
            )
        ).scalars().all()
        account = (
            await session.execute(
                text(
                    "SELECT account_no, cash_balance, available_cash FROM fin_sim_account "
                    "WHERE customer_id = :u"
                ),
                {"u": customer_id},
            )
        ).mappings().one_or_none()
        print(f"\n  客户 {customer_id}（{user['username']}，{user['user_type']}，"
              f"{user['status']}）角色={list(roles) or '无（会 403）'}")
        print(f"    账户：{dict(account) if account else '（无）'}")

        holdings = (
            await session.execute(
                text(
                    "SELECT h.trade_account, p.product_code, p.product_name, h.total_quantity, "
                    "h.available_quantity, h.average_cost, h.cost_amount, h.market_value, "
                    "h.current_value, h.status "
                    "FROM fin_holding h JOIN fin_product p ON p.id = h.product_id "
                    "WHERE h.customer_id = :u ORDER BY p.product_code"
                ),
                {"u": customer_id},
            )
        ).mappings().all()
        if not holdings:
            print("    持仓：（无）")
        for holding in holdings:
            print(
                f"    持仓 {holding['product_code']} {holding['product_name']}："
                f"数量={holding['total_quantity']} 可用={holding['available_quantity']} "
                f"成本={holding['average_cost']} 成本额={holding['cost_amount']} "
                f"市值={holding['market_value']} 状态={holding['status']}"
            )
            if latest_nav is not None and holding["product_code"] == code:
                expected = (Decimal(str(holding["total_quantity"])) * latest_nav).quantize(
                    Decimal("0.01")
                )
                ok = Decimal(str(holding["market_value"])) == expected
                print(f"      [一致性] 市值应为 数量×最新净值 = {expected} -> "
                      f"{'一致' if ok else '不一致'}")


# ------------------------------------------------------------------ 主流程

async def run(args: argparse.Namespace) -> int:
    customers = list(args.customers)
    nav = Decimal(args.nav).quantize(Decimal("0.000001"))
    quantity = Decimal(args.quantity)

    async with SessionFactory() as session:
        if args.show:
            await show_summary(session, args.product_code, customers)
            return 0

        async with session.begin():
            product_id, created = await ensure_product(
                session,
                args.product_code,
                name=args.product_name or f"{args.product_code} 自定义产品",
                exchange=args.exchange,
                category=args.category,
                risk_level=args.risk_level,
                nav=nav,
            )
            print(f"[产品] {args.product_code} -> id={product_id} "
                  f"（{'新建' if created else '已存在，已刷新净值'}），最新净值 {nav}")

            added = await ensure_nav_history(session, product_id, nav, args.days)
            print(f"[净值] 新增 {added} 条历史净值（目标 {args.days} 个交易日）")

            price_state = await ensure_market_price(session, product_id, nav)
            print(f"[行情] {price_state}")

            print()
            for customer_id in customers:
                user_state, role_state = await ensure_user(
                    session,
                    customer_id,
                    # 新建用户必须有密码，否则 password_hash 是占位符、根本登不进来；
                    # 已存在的用户则**只有显式传了 --password 才改**，绝不静默重置。
                    password=args.password or DEFAULT_PASSWORD,
                    password_changed=args.password is not None,
                )
                account_no, cash, account_created = await ensure_account(session, customer_id)
                holding_state, held, market_value = await ensure_holding(
                    session,
                    customer_id,
                    account_no,
                    product_id,
                    quantity=quantity,
                    nav=nav,
                )
                print(f"[客户 {customer_id}] 用户：{user_state}；{role_state}")
                print(f"            账户：{account_no}"
                      f"（{'新建，初始资金 ' + str(INITIAL_BALANCE) if account_created else '已存在'}，"
                      f"现金 {cash}）")
                print(f"            持仓：{args.product_code} {holding_state}，"
                      f"数量 {held}，市值 {market_value}")

        await show_summary(session, args.product_code, customers)

    # ---- 开户风险测评 ----
    # ⚠️ 必须放在上面的事务**提交之后**：接口要能读到刚建的用户才行。
    # 少了这一步，客户登录是成功的，但访问任何非 onboarding 接口都会
    # 403「请先完成开户风险测评问卷」—— 看起来像权限没配好，其实是合规前置没过。
    print("\n===== 开户风险测评（走接口，会一并生成画像快照与标签）=====")
    if args.skip_onboarding:
        print("  [跳过] --skip-onboarding：客户登录后会 403「请先完成开户风险测评问卷」")
    else:
        password = args.password or DEFAULT_PASSWORD
        failed = 0
        for customer_id in customers:
            ok, message = complete_onboarding(args.base_url, customer_id, password)
            print(f"  [{'完成' if ok else '警告'}] 客户 {customer_id}：{message}")
            failed += 0 if ok else 1
        if failed:
            print(f"  ⚠️ {failed} 个客户没完成测评。若服务未启动，先跑 start.ps1，"
                  f"再重跑本脚本；也可用 --skip-onboarding 明确跳过。")

        # 风险等级从库里读回 —— 接口不返回它，凭算容易算错。
        # 这一栏是判断"能不能买这只产品"的依据：等级要 ≥ 产品风险等级。
        print("\n  实际等级（从 fin_risk_assessment 读回）：")
        async with SessionFactory() as session:
            for customer_id in customers:
                row = (
                    await session.execute(
                        text(
                            "SELECT investor_type, total_score, questionnaire_version, valid_until "
                            "FROM fin_risk_assessment WHERE customer_id = :c "
                            "ORDER BY assessed_at DESC LIMIT 1"
                        ),
                        {"c": customer_id},
                    )
                ).mappings().one_or_none()
                if row is None:
                    print(f"    客户 {customer_id}：（没有测评记录）")
                else:
                    print(f"    客户 {customer_id}：{row['investor_type']}  "
                          f"总分 {row['total_score']}  版本 {row['questionnaire_version']}  "
                          f"有效期至 {str(row['valid_until'])[:10]}")

    print("\n客服/前端可直接用下列账号登录查看：")
    for customer_id in customers:
        print(f"  username={customer_id}  密码={args.password or DEFAULT_PASSWORD}")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(
        description="给指定客户建持仓，并保证产品有可查的最新净值",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--product-code", default=DEFAULT_PRODUCT_CODE, help="产品代码")
    parser.add_argument(
        "--customers", type=int, nargs="+", default=list(DEFAULT_CUSTOMERS),
        help="客户账号（同时作为 sys_user.id / username / user_no）",
    )
    parser.add_argument("--nav", default=str(DEFAULT_NAV), help="最新净值（默认 1.0000）")
    parser.add_argument(
        "--quantity", default=str(DEFAULT_QUANTITY), help="每户持仓数量（默认 1000）"
    )
    parser.add_argument("--days", type=int, default=DEFAULT_DAYS, help="造多少个交易日的净值")
    parser.add_argument("--product-name", default=None, help="产品名称（默认按代码生成）")
    parser.add_argument("--exchange", default="SZSE", help="交易所：SZSE / SSE")
    parser.add_argument("--category", default="ETF", help="产品类别")
    parser.add_argument("--risk-level", default="R3", help="风险等级")
    parser.add_argument(
        "--password", default=None,
        help=f"给这些账号设置登录密码（不传则不动已有密码；新建用户用 {DEFAULT_PASSWORD}）",
    )
    parser.add_argument("--show", action="store_true", help="只读：打印现状，不写任何数据")
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="接口地址（用于完成开户测评）")
    parser.add_argument(
        "--skip-onboarding",
        action="store_true",
        help="跳过开户风险测评（跳过则客户访问持仓类接口会 403）",
    )
    args = parser.parse_args()
    return asyncio.run(run(args))


if __name__ == "__main__":
    sys.exit(main())