"""core_tools 查询口径单测（T-11 · FR-C15 + 归零行过滤）。

自建 sqlite 完整种子（`core_product` / `core_holding` / `core_trade`），直调 Tool 函数。
覆盖 T-11 两条 DoD：

1. `query_recent_trades` 的 `sum_amount` 不因 convert 两条流水而**翻倍**
   （走 `amount_view`；明细与 `total_count` 仍为全量），
   并与 `core_ro.sum_trades_on_date` 做**跨口径一致性**断言（自检第 13 问：
   同一口径不得有两份实现漂移）；
2. `query_holdings` 不返回 `qty = 0` 的**归零行**（convert 转出全部份额后的台账留痕行）。

种子时间刻意取**同一个 `now`**：既落在 `query_recent_trades` 的 [now−30d, now) 窗内，
又保证属于 `now.date()` 这一天，使跨口径断言不受运行时刻影响。
"""

from __future__ import annotations

import datetime as _dt
from decimal import Decimal

import pytest
from sqlalchemy import text

from _ddl import create_sqlite_engine, seed_suitability_matrix

from app.gateway.convert_core_repository import ConvertCoreRepository
from app.repository.convert_repository import ConvertRepository
from app.repository.convert_request_repository import ConvertRequestRepository
from app.repository.core_ro import CoreReadOnlyRepository
from app.repository.risk_repository import RiskRepository
from app.service.convert.confirm_service import confirm_one
from app.service.convert.convert_service import accept_convert
from app.service.risk.rules import RiskThresholds
from app.tool.core_tools import query_holdings, query_recent_trades

CUST = "CUST-T11"


@pytest.fixture()
def seed():
    engine = create_sqlite_engine()
    now = _dt.datetime.now()
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type,"
                " min_subscribe_amount, term_days) VALUES"
                " ('PA', '甲基金', 'R3', 'mixed', 100, 0),"
                " ('PB', '乙基金', 'R3', 'mixed', 100, 0)"
            )
        )
        # PA 正常持有；PB 为 convert 转出全部后的**归零行**（qty = 0，行保留）
        conn.execute(
            text(
                "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount,"
                " market_value, pnl_pct, as_of) VALUES"
                " (:cid, 'PA', 1000.00, 1000.00, 1200.00, 0.2000, :d),"
                " (:cid, 'PB', 0.00, 0.00, 0.00, 0.0000, :d)"
            ),
            {"cid": CUST, "d": now.date()},
        )
        # 一次 convert 落两条（转出 redeem 300000 + 转入 subscribe 300000，共享 G1）
        # + 1 笔普通赎回 100000（convert_group_id = NULL）
        conn.execute(
            text(
                "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount,"
                " qty, convert_group_id, trade_status, traded_at) VALUES"
                " ('TX-OUT',   :cid, 'PA', 'redeem',    300000, 250.00, 'G1', 'confirmed', :t),"
                " ('TX-IN',    :cid, 'PB', 'subscribe', 300000, 240.00, 'G1', 'confirmed', :t),"
                " ('TX-PLAIN', :cid, 'PA', 'redeem',    100000,  80.00, NULL, 'confirmed', :t)"
            ),
            {"cid": CUST, "t": now},
        )
    yield engine
    engine.dispose()


def test_query_recent_trades_sum_not_doubled(seed):
    """FR-C15：汇总只计转出端；**明细与条数保持全量**（一次转换两条是真实的）。"""
    repo = CoreReadOnlyRepository(engine=seed)
    res = query_recent_trades(CUST, days=30, core_ro=repo)

    # 明细全量：convert 两条 + 普通赎回 1 条
    assert res["total_count"] == 3
    assert sorted(r["trade_type"] for r in res["items"]) == ["redeem", "redeem", "subscribe"]

    # 汇总去重：300000（转出端）+ 100000（普通赎回）= 400000
    # 若未去重则为 700000（TX-OUT + TX-IN 双计）—— 用具体值才区分得开
    assert res["sum_amount"] == 400000.0


def test_query_recent_trades_sum_matches_sum_trades_on_date(seed):
    """跨口径一致性：Tool 汇总（Python 侧 amount_view）== 仓储 SQL 汇总（IS NULL OR =''）。

    两处是同一口径的两种落地（一个 Python 一个 SQL），此处用断言锁死等价性，
    任一侧口径漂移都会在此变红。
    """
    repo = CoreReadOnlyRepository(engine=seed)
    tool_sum = query_recent_trades(CUST, days=30, core_ro=repo)["sum_amount"]
    sql_sum = repo.sum_trades_on_date(CUST, _dt.date.today())
    assert Decimal(str(tool_sum)) == sql_sum
    assert sql_sum == Decimal("400000")


def test_query_holdings_excludes_zero_qty(seed):
    """归零行（convert 转出全部份额，`qty = 0` 台账留痕）不得作为持仓返回。"""
    repo = CoreReadOnlyRepository(engine=seed)
    res = query_holdings(CUST, core_ro=repo)

    assert res["total_count"] == 1
    assert [r["product_id"] for r in res["items"]] == ["PA"]
    # 合计不含归零行（PB 的 market_value 为 0，即使计入也不变；
    # 故断言**条数**才是真正能区分对错的判据）
    assert res["sum_market_value"] == 1200.0


# ── T+1 确认段场景（v2.0 T-11 · 开发计划 R-回归 8 / R-回归 13 的补用例项）────────
# 既有 3 条用例以直接插 SQL 的方式模拟 convert 流水，锁定**口径本身**；
# 下面两条改走真实链路（accept_convert → confirm_one），锁定「确认事务真实落库
# 之后」口径依然成立 —— T+1 模型下流水与持仓**只在确认段产生**，这是两条新用例
# 相对既有用例的独立价值（不是重复覆盖）：
#   ① 确认落两条同 gid 流水 → sum_amount 只计转出端一次（验收 11）
#   ② 全转归零 → core_holding.qty=0 行保留（台账留痕）但不在持仓查询结果（F-12）

PROD_T11_OUT = "PROD-T11-OUT"
PROD_T11_IN = "PROD-T11-IN"
COMPANY = "华夏模拟基金"
TA = "TA-CN-001"

T_DAY = _dt.date(2026, 9, 4)  # 受理日 T（周五）
T1_DAY = _dt.date(2026, 9, 7)  # 确认业务日 T+1（下周一）
SUBMIT_AT = _dt.datetime(2026, 9, 4, 10, 0)
CONFIRM_AT = _dt.datetime(2026, 9, 7, 9, 0)

OUT_NAV = Decimal("1.3604")
IN_NAV = Decimal("1.9194")
FEE_TIERS = [
    (0, 7, "0.0150"), (7, 30, "0.0100"), (30, 180, "0.0050"),
    (180, 365, "0.0025"), (365, None, "0.0000"),
]
# 全转 30000 份的折算锚点：转出金额 30000×1.3604 = 40812.00（费前，与 confirm 返回
# out_amount 同源），持有期落 (30,180) 档 → 赎回费 40812.00×0.50% = 204.06 单列 redeem_fee。
# core_trade 转出端流水 amount = 费前 40812.00（R-b 口径，与 PRD §5.3.2 示例 68020.00 同语义）。
OUT_AMOUNT = Decimal("40812.00")
REDEEM_FEE = Decimal("204.06")
PLAIN_AMOUNT = Decimal("100000")


@pytest.fixture()
def t1_env():
    """T+1 链路最小种子：C3 客户、转出 R2 债基（1 批 30000 份 + 持仓 30000）、转入 R4 股基。

    两端同管理人同 TA（受理段硬约束）；批次 confirmed_at=2026-07-01 → 持有期落
    (30,180) 档 0.50%，赎回费数值确定（见 OUT_AMOUNT 注释），断言可逐字节对账。
    """
    engine = create_sqlite_engine()
    seed_suitability_matrix(engine)  # C×R 矩阵是 check_suitability 的 L0 权威，缺失即全量 forbidden
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO core_customer (customer_id, display_name, age, is_active)"
                " VALUES (:c, 'T11客户', 40, 1)"
            ),
            {"c": CUST},
        )
        conn.execute(
            text(
                "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at, expires_at)"
                " VALUES (:c, 'C3', :t, :e)"  # C3 → 转入 R4 放行；有效期覆盖 T 与 T+1（D25 复核）
            ),
            {"c": CUST, "t": SUBMIT_AT - _dt.timedelta(days=30), "e": SUBMIT_AT + _dt.timedelta(days=300)},
        )
        conn.execute(
            text(
                "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type,"
                " can_subscribe, can_redeem, subscribe_fee_rate, min_redeem_qty, min_hold_qty,"
                " fund_company, ta_code) VALUES"
                " (:po, 'T11转出债基', 'R2', 'bond', 1, 1, 0.0030, 0, 0, :co, :ta),"
                " (:pi, 'T11转入股基', 'R4', 'stock', 1, 1, 0.0080, 0, 0, :co, :ta)"
            ),
            {"po": PROD_T11_OUT, "pi": PROD_T11_IN, "co": COMPANY, "ta": TA},
        )
        for lo, hi, rate in FEE_TIERS:
            conn.execute(
                text(
                    "INSERT INTO core_fee_rule (product_id, fee_type, min_hold_days,"
                    " max_hold_days, rate) VALUES (:p, 'redeem', :a, :b, :r)"
                ),
                {"p": PROD_T11_OUT, "a": lo, "b": hi, "r": float(rate)},
            )
        conn.execute(
            text(
                "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) VALUES"
                " (:po, :no, 0, :d), (:pi, :ni, 0, :d)"
            ),
            {"po": PROD_T11_OUT, "no": float(OUT_NAV), "pi": PROD_T11_IN, "ni": float(IN_NAV), "d": T_DAY},
        )
        conn.execute(
            text(
                "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, remain_qty,"
                " nav, confirmed_at) VALUES (:l, :c, :p, 30000, 30000, 1.0300, :cat)"
            ),
            {"l": "LOT-T11-1", "c": CUST, "p": PROD_T11_OUT, "cat": _dt.datetime(2026, 7, 1, 10, 0, 0)},
        )
        conn.execute(
            text(
                "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount,"
                " market_value, pnl_pct, as_of) VALUES (:c, :p, 30000, 30000, 40812.0, 0, :d)"
            ),
            {"c": CUST, "p": PROD_T11_OUT, "d": T_DAY},
        )
        cursor = _dt.date(2026, 9, 1)
        for _ in range(60):
            if cursor.weekday() < 5:  # 周一~周五开市；周末不插行 = 休市
                conn.execute(
                    text("INSERT INTO core_trade_calendar (cal_date, is_open, remark) VALUES (:d, 1, 't')"),
                    {"d": cursor},
                )
            cursor += _dt.timedelta(days=1)
    yield engine
    engine.dispose()


def _t11_services(engine) -> dict:
    return dict(
        core_ro=CoreReadOnlyRepository(engine=engine),
        risk_repo=RiskRepository(engine=engine),
        convert_repo=ConvertRepository(engine=engine),
        request_repo=ConvertRequestRepository(engine=engine),
    )


def _quiet_th() -> RiskThresholds:
    """全阈值推到不可达：本组用例只关心 core_tools 查询口径，不让引擎出单添噪。"""
    big = Decimal("999999999")
    return RiskThresholds(
        large_amount=big, daily_total=big, freq_count=999, probe_window_minutes=5,
        probe_count=999, probe_amount=big, small_amount=Decimal("0.01"), small_count=999,
        concentration_threshold=1.01,
    )


def _accept_full(engine) -> dict:
    """受理**全转**（qty = 批次总量 30000）：确认后转出端持仓必然归零（F-12 场景）。"""
    return accept_convert(
        {
            "customer_id": CUST,
            "from_product_id": PROD_T11_OUT,
            "to_product_id": PROD_T11_IN,
            "qty": Decimal("30000"),
            "client_request_id": "T11-REQ-1",
        },
        now=SUBMIT_AT,
        **_t11_services(engine),
    )


def _confirm(engine, gid: str) -> dict:
    return confirm_one(
        gid,
        now=CONFIRM_AT,
        as_of=T1_DAY,
        core_writer=ConvertCoreRepository(engine=engine),
        thresholds=_quiet_th(),
        **_t11_services(engine),
    )


def _t11_rows(engine, sql: str, **params) -> list[dict]:
    with engine.connect() as conn:
        return [dict(r) for r in conn.execute(text(sql), params).mappings()]


def test_confirm_then_sum_amount_counts_once(t1_env):
    """验收 11（确认段场景）：真实受理→确认落两条同 gid 流水后，sum_amount 只计转出端。"""
    engine = t1_env
    gid = _accept_full(engine)["convert_group_id"]
    res = _confirm(engine, gid)
    assert res["status"] == "confirmed"
    assert res["out_amount"] == "40812.00"  # 折算锚点逐字节对账（30000×1.3604，费前）
    assert res["redeem_fee"] == "204.06"  # 持有期落 (30,180) 档 0.50% 的证明

    # 同一确认日再落一笔无组普通赎回（无 convert_group_id → 金额全额计入）
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount,"
                " qty, convert_group_id, trade_status, traded_at) VALUES"
                " ('TRD-T11-PLAIN', :cid, :pid, 'redeem', 100000.0, 80.0, NULL, 'confirmed', :t)"
            ),
            {"cid": CUST, "pid": PROD_T11_OUT, "t": CONFIRM_AT},
        )

    repo = CoreReadOnlyRepository(engine=engine)
    out = query_recent_trades(CUST, days=30, core_ro=repo)

    # 明细全量：convert 两条（redeem+subscribe）+ 普通赎回 1 条 —— 一次转换两条是真实的
    assert out["total_count"] == 3
    assert sorted(r["trade_type"] for r in out["items"]) == ["redeem", "redeem", "subscribe"]

    # 汇总只计转出端：40812.00 + 100000 = 140812.00；
    # 转入端 in_amount 若被计入即翻倍（差额恰为一条转入流水）—— 用具体值才区分得开
    assert Decimal(str(out["sum_amount"])) == OUT_AMOUNT + PLAIN_AMOUNT

    # 跨口径一致性（T+1 确认场景）：SQL 侧同日累计与 Tool 汇总一致（流水 traded_at = T+1）
    assert repo.sum_trades_on_date(CUST, T1_DAY) == OUT_AMOUNT + PLAIN_AMOUNT


def test_confirm_full_transfer_holding_zero_row_excluded(t1_env):
    """F-12 / R-回归 13（确认段场景）：全转后 `core_holding.qty=0` 行保留但不是持仓。"""
    engine = t1_env
    gid = _accept_full(engine)["convert_group_id"]
    res = _confirm(engine, gid)
    assert res["status"] == "confirmed"

    # 台账留痕行仍在（D 决策：转出归零保留 qty=0 行，不物理删除）——
    # 先证明行存在，后面的「不在持仓结果」断言才有区分度（否则可能是行被删了）
    rows = _t11_rows(
        engine,
        "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUST, p=PROD_T11_OUT,
    )
    assert len(rows) == 1 and Decimal(str(rows[0]["qty"])) == 0

    repo = CoreReadOnlyRepository(engine=engine)
    # SQL 层（core_ro.list_holdings，`qty > 0` 过滤）：只剩转入端持仓
    listed = repo.list_holdings(CUST)
    assert [r["product_id"] for r in listed] == [PROD_T11_IN]
    assert Decimal(str(listed[0]["qty"])) == Decimal(res["in_qty"])  # 确认段两端持仓同事务落对

    # Tool 层（core_tools.query_holdings）与 SQL 层同口径
    holdings = query_holdings(CUST, core_ro=repo)
    assert holdings["total_count"] == 1
    assert [r["product_id"] for r in holdings["items"]] == [PROD_T11_IN]