# -*- coding: utf-8 -*-
"""
上游榜单的版本比对 (纯逻辑, 零外部依赖)
========================================
上游 `/plan` 应答里有个 `changes` 字段, 本该回答「跟上一份比, 谁新进榜、谁掉榜」。
实测它**恒为 `null`** (UPSTREAM_PLAN_API.md §4 Q4: 已问, 未给非空示例)。等它补, 就是把
一件 PMS 自己能算的事挂在外部依赖上 —— 于是改成 PMS 每次拉计划落一份名册快照, 差异自己算。

顺带补上 §7.3 那个洞: `/plan` 是 `plan.collect()` **实时算**的, 同一个 `date` 可以对应
多个版本 (07-31 实测: 相隔一小时的两次请求, 打分池 972/81 → 423/502, 榜首换人), 而应答里
没有 `generated_at` 或版本戳 —— **下游此前没有任何办法判断"手上这份是哪一版"**。落了快照,
这个问题就地解决: 同一 `plan_date` 底下有几行, 就是上游那天重算过几版。

三种比对语义, 混在一起看会得出错误结论, 所以显式分开 (`kind`):

  * `first`               —— 没有上一份快照。**一条变化都不报** (否则首次全表 300 条"新进榜")
  * `cross_day`           —— 计划日不同。这是 Q4 说的「每日几十只」的正常升降档
  * `same_date_revision`  —— **同一个计划日的不同版本**。这不是市场变化, 是上游重算;
                             它意味着 PMS 一小时前拿到的候选池与现在不是一回事, 要显眼报出来

--------------------------------------------------------------------------------
掉榜为什么不能直接算「上一版有、这一版没有」
--------------------------------------------------------------------------------
我们向上游要 `top=300`。一只票从 rank 250 掉到 rank 310, 它会从名册里"消失"—— 但这不是
上游观点变化, 只是排到了我们要的条数之外。**吃满 top 的那一版, 尾部的进出全是截断噪音。**
不作区分的话, 每天会多出几十上百条假掉榜, 提示就没人看了 (同 `_capped` 治的那个假警报,
是同一类错误的两个面)。

新进榜是**对称**的同一个问题: 上一版吃满了 top, 那么这一版排在末尾的"新面孔"很可能上一版
就在榜上, 只是排在了 300 名开外看不见。所以两道闸各看各的那一版:

    掉榜 —— 看 **这一版** 吃没吃满 (它截断了, 掉出去的可能只是被截掉)
    新进 —— 看 **上一版** 吃没吃满 (它截断了, 新面孔可能上一版就在, 只是没露面)

`tail_guard` 是那道闸的位置: 排名进入榜单前 `tail_guard` 比例 (默认一半) 才算数, 其余计进
`tail_churn` 只报个数、不列名。没吃满 top 的那一版不设闸 —— 那种情况下的进出是真的。

**持仓票不受这道闸约束。** 闸是拿"少报几条噪音"换"提示还有人看", 这笔账在持仓票上不成立:
非持仓票误报一条, 代价是白看一眼; **持仓票漏报一条, 代价是一个该减没减的仓位**。所以持仓票
一律照报, 并在 `tail_churn.held_exempt` 里记下有几条是靠这条豁免出来的。
"""
from __future__ import annotations

import hashlib

# 传导档位的强弱次序 (上游 2026-07-31 答复 Q7: 强传导 > 弱传导 > 无传导)。
# 不在表里的档位 **不参与升降判定** —— 上游哪天加了新档位, 宁可不报, 不能报反。
TIER_ORDER = {"强传导": 3, "弱传导": 2, "无传导": 1}

BUCKET_MAIN, BUCKET_OBSERVE = "main", "observe"

KIND_FIRST, KIND_CROSS_DAY, KIND_REVISION = "first", "cross_day", "same_date_revision"

DEFAULT_RANK_JUMP = 50      # 名次跳变多少名才值得报
DEFAULT_TAIL_GUARD = 0.5    # 吃满 top 时, 上一版排名在前多少比例内的掉榜才算数

# 名册里每条存哪些字段。键名取短的 —— 一份名册 300~400 条, 要落进 MEDIUMTEXT。
_FIELDS = ("c", "n", "r", "s", "t", "h", "b")


def tier_rank(tier):
    """档位 → 强弱序数; 未知档位返回 None (判不了升降就不判)。"""
    if tier is None:
        return None
    return TIER_ORDER.get(str(tier).strip())


# ================================================================ 名册
def roster_of(plan: dict) -> dict:
    """`plan_feed.parse_plan()` 的结果 → 名册 {ts_code: 精简行}。

    主榜在前, 同一只票若两档都出现以主榜为准 (与 parse_plan 里 themes 的口径一致)。
    """
    out = {}
    for bucket in (BUCKET_MAIN, BUCKET_OBSERVE):
        for r in (plan.get(bucket) or []):
            code = r.get("ts_code")
            if not code or code in out:
                continue
            out[code] = {"c": code, "n": r.get("name"), "r": r.get("rank"),
                         "s": r.get("score"), "t": r.get("tier"),
                         "h": r.get("theme"), "b": r.get("bucket") or bucket}
    return out


def roster_rows(roster: dict) -> list:
    """名册 → 落库用的 list (代码序, 保证同一份名册序列化结果稳定)。"""
    return [{k: roster[c].get(k) for k in _FIELDS} for c in sorted(roster)]


def roster_from_rows(rows) -> dict:
    """落库 list → 名册 dict。坏行跳过, 不让一条脏数据废掉整份快照。"""
    out = {}
    for r in (rows or []):
        if not isinstance(r, dict):
            continue
        code = r.get("c") or r.get("ts_code")
        if not code:
            continue
        out[code] = {"c": code, "n": r.get("n"), "r": r.get("r"), "s": r.get("s"),
                     "t": r.get("t"), "h": r.get("h"), "b": r.get("b")}
    return out


def digest_of(roster: dict) -> str:
    """名册指纹 —— 同一份榜重复拉不重复落。

    **不含 score。** `/plan` 是实时算的, score 末位天天抖; 把它算进指纹, 每次缓存过期重拉
    都会多落一行快照, 表白涨而信息量为零。含 rank 就够了: 分数抖到改变了次序才算真的变了,
    没改变次序的抖动对候选池没有任何影响 (候选就是按 score 降序切前 N)。
    """
    h = hashlib.sha1()
    for c in sorted(roster):
        r = roster[c]
        h.update(("%s|%s|%s|%s|%s\n" % (c, r.get("r"), r.get("t") or "",
                                        r.get("h") or "", r.get("b") or "")).encode("utf-8"))
    return h.hexdigest()


# ================================================================ 比对
def _row(r: dict, **extra) -> dict:
    out = {"ts_code": r.get("c"), "name": r.get("n"), "rank": r.get("r"),
           "score": r.get("s"), "tier": r.get("t"), "theme": r.get("h"),
           "bucket": r.get("b")}
    out.update(extra)
    return out


def _bucket_len(roster: dict, bucket: str) -> int:
    return sum(1 for r in (roster or {}).values() if (r.get("b") or BUCKET_MAIN) == bucket)


def _cutoff(roster: dict, capped, ratio: float) -> dict:
    """尾部闸的位置 {bucket: 名次上限 or None}。

    只有那一版**吃满了 top** 才设闸 —— 没吃满说明上游能给的都给了, 榜尾的进出是真的。
    `capped` 按档给 (`{"main": bool, "observe": bool}`): 两档是两个独立的请求参数
    (`top` / `obs_top`), 主榜吃满**完全不意味着**观察档也吃满了。早先只传主榜那一个标志,
    效果是拿主榜的截断去解释观察档的进出 —— 观察档一次真实的摘牌会被记成"截断噪音"。
    该档一条都没有时也不设闸: 整档消失/整档出现本身就是要报的大事, 不该被闸吞掉。
    """
    caps = capped if isinstance(capped, dict) else {BUCKET_MAIN: bool(capped),
                                                    BUCKET_OBSERVE: bool(capped)}
    out = {}
    for b in (BUCKET_MAIN, BUCKET_OBSERVE):
        n = _bucket_len(roster, b)
        out[b] = (n * ratio) if (caps.get(b) and n > 0) else None
    return out


def _tail_dropped(row: dict, rank, cut: dict) -> bool:
    """这条进/出是不是只是榜尾截断的噪音。

    rank 缺失时返回 False = **照报**。全模块的口径是"判不了就不判"—— 在别处那意味着不报
    变化, 在这里却要反过来: 这个函数的"是"代表**吞掉一条变化**, 所以拿不准时必须放行。
    """
    lim = cut.get(row.get("b") or BUCKET_MAIN)
    return lim is not None and isinstance(rank, int) and rank > lim


def diff(prev_roster, curr_roster, *, prev_date=None, curr_date=None, held=(),
         rank_jump: int = DEFAULT_RANK_JUMP, tail_guard: float = DEFAULT_TAIL_GUARD,
         prev_capped=False, curr_capped=False, prev_broken: bool = False) -> dict:
    """两份名册的差异。**不抛错** —— 页面提示不该有能力搞崩取数。

    held: 当前持仓代码 (点式, 调用方负责归一)。持仓票的变化单独拎出来, 且**不受尾部闸约束**
    —— 上游把一只持仓票摘出榜或降了档, 是"该不该继续拿着"的直接信号, 漏报的代价比误报大
    得多 (见模块头部)。

    prev_capped / curr_capped: 那一版吃没吃满请求的条数。可以给 bool, 也可以按档给
    `{"main": ..., "observe": ...}` —— 两档是两个独立的请求参数, 主榜吃满不代表观察档也满。
    掉榜看 `curr_capped`、新进看 `prev_capped`, 各看各的那一版, 理由见模块头部。

    prev_broken: 上一版**存在但读不出来** (名册落库时坏了)。这跟"没有上一版"是两回事:
    后者是正常的首次, 前者是故障 —— 必须说成故障, 否则一条真实变化都不报还显示得很正常。
    """
    held = {c for c in (held or []) if c}
    curr_roster = curr_roster or {}
    if not prev_roster:
        note = ("上一版名册读不出来 (快照损坏?) —— **这次比不了**, 下面的空白不代表没有变化"
                if prev_broken else
                "首次落快照, 无可比对的上一版 —— 变化提示从下一次拉取开始")
        return _empty(KIND_FIRST, prev_date, curr_date, curr_roster, held, note=note,
                      rank_jump=rank_jump, tail_guard=tail_guard, prev_broken=prev_broken)

    kind = (KIND_REVISION if (prev_date and curr_date and prev_date == curr_date)
            else KIND_CROSS_DAY)

    entered, exited = [], []
    tail_churn = {"entered": 0, "exited": 0, "held_exempt": 0}
    tier_up, tier_down, bucket_moved, jumps = [], [], [], []

    # 两道尾部闸各看各的那一版 —— 见模块头部
    ratio = max(0.0, min(1.0, float(tail_guard if tail_guard is not None else DEFAULT_TAIL_GUARD)))
    exit_cut = _cutoff(curr_roster, curr_capped, ratio)    # 掉榜: 这一版截没截
    enter_cut = _cutoff(prev_roster, prev_capped, ratio)   # 新进: 上一版截没截

    for code, p in (prev_roster or {}).items():
        if curr_roster.get(code) is not None:
            continue
        h = code in held
        if _tail_dropped(p, p.get("r"), exit_cut):
            if not h:
                tail_churn["exited"] += 1   # 本来就在榜尾 —— 大概率只是被这一版的 top 截掉
                continue
            tail_churn["held_exempt"] += 1  # 持仓票不适用这道闸: 漏报一条就是一个仓位
        exited.append(_row(p, held=h))

    for code, c in curr_roster.items():
        p = (prev_roster or {}).get(code)
        if p is None:
            h = code in held
            if _tail_dropped(c, c.get("r"), enter_cut):
                if not h:
                    tail_churn["entered"] += 1   # 上一版就吃满了, 这张脸当时可能只是没露面
                    continue
                tail_churn["held_exempt"] += 1
            entered.append(_row(c, held=h))
            continue
        h = code in held
        pb, cb = p.get("b") or BUCKET_MAIN, c.get("b") or BUCKET_MAIN
        if pb != cb:
            bucket_moved.append(_row(c, held=h, moved_from=pb, moved_to=cb))
        pt, ct = tier_rank(p.get("t")), tier_rank(c.get("t"))
        if pt is not None and ct is not None and pt != ct:
            (tier_up if ct > pt else tier_down).append(
                _row(c, held=h, tier_from=p.get("t"), tier_to=c.get("t")))
        pr, cr = p.get("r"), c.get("r")
        if pb == cb and isinstance(pr, int) and isinstance(cr, int):
            d = cr - pr
            if abs(d) >= max(1, int(rank_jump or DEFAULT_RANK_JUMP)):
                jumps.append(_row(c, held=h, rank_from=pr, rank_to=cr, rank_delta=d))

    entered.sort(key=lambda x: (x["rank"] if isinstance(x["rank"], int) else 10 ** 9))
    exited.sort(key=lambda x: (x["rank"] if isinstance(x["rank"], int) else 10 ** 9))
    jumps.sort(key=lambda x: abs(x.get("rank_delta") or 0), reverse=True)
    for lst in (tier_up, tier_down, bucket_moved):
        lst.sort(key=lambda x: (x["rank"] if isinstance(x["rank"], int) else 10 ** 9))

    out = {"kind": kind, "prev_date": prev_date, "curr_date": curr_date,
           "entered": entered, "exited": exited, "tier_up": tier_up, "tier_down": tier_down,
           "bucket_moved": bucket_moved, "rank_jump": jumps, "tail_churn": tail_churn,
           "tail_guarded": {"entered": _guard_view(enter_cut), "exited": _guard_view(exit_cut)},
           "roster_size": {"prev": len(prev_roster or {}), "curr": len(curr_roster)},
           "prev_broken": False, "date_regressed": _regressed(prev_date, curr_date),
           "params": {"rank_jump": int(rank_jump or DEFAULT_RANK_JUMP), "tail_guard": ratio}}
    out["counts"] = {k: len(out[k]) for k in
                     ("entered", "exited", "tier_up", "tier_down", "bucket_moved", "rank_jump")}
    out["held"] = _held_view(out, held)
    out["note"] = _note(out)
    return out


def _guard_view(cut: dict) -> dict:
    """哪几档真的设了闸, 闸位在第几名 —— 让"为什么这条没报"可查, 而不是只能猜。"""
    return {b: (None if cut.get(b) is None else round(cut[b], 1))
            for b in (BUCKET_MAIN, BUCKET_OBSERVE)}


def _regressed(prev_date, curr_date) -> bool:
    """这一版的计划日比上一版还早。

    正常流程不会这样 —— 快照按落库次序排, 而计划日是往前走的。会出现只有一种情况:
    有人拿 `--date` 显式拉了一份旧计划。那时"新进/掉榜"的方向是反的, 得说明白。
    """
    return bool(prev_date and curr_date and str(curr_date) < str(prev_date))


def _empty(kind, prev_date, curr_date, curr_roster, held, note="",
           rank_jump=DEFAULT_RANK_JUMP, tail_guard=DEFAULT_TAIL_GUARD,
           prev_broken=False) -> dict:
    ratio = max(0.0, min(1.0, float(tail_guard if tail_guard is not None else DEFAULT_TAIL_GUARD)))
    out = {"kind": kind, "prev_date": prev_date, "curr_date": curr_date,
           "entered": [], "exited": [], "tier_up": [], "tier_down": [],
           "bucket_moved": [], "rank_jump": [],
           "tail_churn": {"entered": 0, "exited": 0, "held_exempt": 0},
           "tail_guarded": {"entered": _guard_view({}), "exited": _guard_view({})},
           "roster_size": {"prev": 0, "curr": len(curr_roster or {})},
           "prev_broken": bool(prev_broken), "date_regressed": _regressed(prev_date, curr_date),
           # 回显**调用方实际给的**参数, 不是默认值 —— 页面拿它显示当前口径, 写死会骗人
           "params": {"rank_jump": int(rank_jump or DEFAULT_RANK_JUMP), "tail_guard": ratio}}
    out["counts"] = {k: 0 for k in
                     ("entered", "exited", "tier_up", "tier_down", "bucket_moved", "rank_jump")}
    out["held"] = _held_view(out, held)
    out["note"] = note
    return out


def _held_view(d: dict, held) -> dict:
    """持仓票命中的变化。**掉榜与降档排在最前** —— 那是要不要继续持有的信号。

    `n_watch` 是"要盯的"那几条, 页面红条与探活脚本都拿它当判据。三类算进去:

      掉榜 · 降档 · **主榜→观察档**

    第三类容易漏。观察档的定义是"无券商覆盖", 所以掉进观察档的含义是**估值锚没了**, 与
    降档是同一量级的信号; 而观察档的行没有 `tier`, 于是它既不算 `exited` 也不算
    `tier_down` —— 不显式算进来的话, 一只持仓票丢了券商覆盖, 页面红条不亮、探活脚本还会
    打一句"持仓票没被摘也没降档"。
    """
    pick = lambda k: [x for x in d.get(k) or [] if x.get("held")]   # noqa: E731
    out = {k: pick(k) for k in ("exited", "tier_down", "bucket_moved",
                                "tier_up", "entered", "rank_jump")}
    out["coverage_lost"] = [x for x in out["bucket_moved"] if x.get("moved_to") == BUCKET_OBSERVE]
    out["n_watch"] = len(out["exited"]) + len(out["tier_down"]) + len(out["coverage_lost"])
    out["n_total"] = sum(len(v) for k, v in out.items()
                         if isinstance(v, list) and k != "coverage_lost")   # 别重复计 bucket_moved
    return out


def _note(d: dict) -> str:
    """一句人话结论 —— 页面横幅与日志直接用这句, 不必各写一份措辞。"""
    c, h = d["counts"], d["held"]
    if d["kind"] == KIND_REVISION:
        head = (f"同一计划日 {d['curr_date']} 的**新版本**: 上游重算过 —— "
                f"此前拿到的候选池与现在不是同一份")
    else:
        head = f"计划日 {d['prev_date']} → {d['curr_date']}"
    body = (f"新进 {c['entered']} · 掉榜 {c['exited']} · 升档 {c['tier_up']} · "
            f"降档 {c['tier_down']} · 覆盖翻转 {c['bucket_moved']}")
    tail = ""
    if h["n_watch"]:
        tail = f" · **持仓票 {h['n_watch']} 只掉榜/降档/丢了券商覆盖**"
    elif h["n_total"]:
        tail = f" · 持仓票命中 {h['n_total']} 条"
    churn = d.get("tail_churn") or {}
    n_churn = int(churn.get("entered") or 0) + int(churn.get("exited") or 0)
    if n_churn:
        tail += f" · 榜尾进出 {n_churn} 只未计入 (那一版吃满 top, 尾部不可信)"
    if churn.get("held_exempt"):
        tail += f" · 其中 {churn['held_exempt']} 只因是持仓票照报"
    if d.get("date_regressed"):
        tail += " · **注意计划日是往回走的** (拉了一份旧计划?), 新进/掉榜的方向是反的"
    return f"{head}: {body}{tail}"