候选卡判决模块与离线单测；候选卡三张只读视图 SQL（独立文件）；.git 文件解析版本号；坏信号集合桥内归一到 card.py
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
zlt
parent
e6fdb891e8
commit
c540ea1b49
|
|
@ -0,0 +1,145 @@
|
||||||
|
"""候选卡：把一只票的各条证据线装配成判决与理由（纯函数，零 IO）。
|
||||||
|
|
||||||
|
## 为什么要有它
|
||||||
|
|
||||||
|
桥原来的产物是一个分数：主榜分 = 200 + 传导档位×20 + 组内分。分数排在最前的
|
||||||
|
强传导档，实证里过去一个月是四档中最差的（五日超额 −2.04，见
|
||||||
|
docs/主观选股改进方案_2026-09-02.md 第 1.8b 节）。方案把桥从"打分排序器"改成
|
||||||
|
"候选卡装配器"：分数与档位一律不动，另出一份带理由的候选单。规则只有三条硬门槛、
|
||||||
|
一条确认、三条硬风险，全部是一个主观选股的人会用的判断，不算总分。
|
||||||
|
|
||||||
|
门槛一 所在环节当日被传导指向（只认环节类目标）
|
||||||
|
门槛二 成员当日已启动：数据日涨幅达到 CARD_START_PCT（默认 3%，与基座热点扫描同口径）
|
||||||
|
门槛三 有券商覆盖、预期空间不低于负容忍线、非 ST 与退市族
|
||||||
|
确认 决策系统吸筹评分为"明确吸筹"，且评分日龄不超过 CARD_ACCUM_MAX_AGE 个交易日
|
||||||
|
硬风险 决策系统昨夜给出卖出、回避或剔除信号；传导快照日与计划日不符；吸筹评分为高位派发
|
||||||
|
|
||||||
|
判决 候选 = 三门槛全过、硬风险为空、确认成立
|
||||||
|
关注 = 硬风险为空，且（只差覆盖一项）或（门槛全过但无确认）
|
||||||
|
仅展示 = 其余，卡上标明未过项
|
||||||
|
|
||||||
|
依据：方案 1.8d——"环节被指向、当日已涨 3% 以上、且明确吸筹"是纯可交易口径下唯一
|
||||||
|
合并样本为正的规则（五日 +2.18、十日 +1.94），但它是顺风策略、不免疫环境；两个旋钮
|
||||||
|
保持默认，等样本外复盘读数再拍，不在历史样本上挑参数（2026-09-02 拍板）。
|
||||||
|
|
||||||
|
## 边界
|
||||||
|
|
||||||
|
本模块不读库、不读配置，只吃调用方装配好的证据字典，输出判决字典。这样它能
|
||||||
|
离线单测（test_card.py），也保证同一段规则在计划装配、复盘脚本、对账工具里只有一份。
|
||||||
|
坏信号集合 BAD_SIGNALS 是"三处同源"纪律里桥的那一份（另两处在决策系统 pms_advisor.py
|
||||||
|
与 PMS rule_gate.py），改一处必须三处同改；pool.py 从这里引用，桥内只此一处。
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
# 决策系统夜间结论表 signal_type 里的坏信号（三处同源，见模块说明）
|
||||||
|
BAD_SIGNALS = {"SELL", "AVOID", "DROPPED"}
|
||||||
|
|
||||||
|
VERDICT_CANDIDATE = "候选"
|
||||||
|
VERDICT_WATCH = "关注"
|
||||||
|
VERDICT_SHOW = "仅展示"
|
||||||
|
_VERDICT_ORDER = {VERDICT_CANDIDATE: 0, VERDICT_WATCH: 1, VERDICT_SHOW: 2}
|
||||||
|
|
||||||
|
|
||||||
|
def _num(v) -> float | None:
|
||||||
|
if v is None:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
x = float(v)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
if x != x: # NaN
|
||||||
|
return None
|
||||||
|
return x
|
||||||
|
|
||||||
|
|
||||||
|
def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
|
||||||
|
accum_max_age: int = 30, neg_tol: float = 0.0) -> dict[str, Any]:
|
||||||
|
"""一只票的证据字典 -> 判决字典。
|
||||||
|
|
||||||
|
ev 里认的键（缺键按缺失处理，不报错）：
|
||||||
|
pointed bool 所在环节当日被传导指向（只认 Segment 目标）
|
||||||
|
theme str 目标环节名；n_sources int 源数；chain_fit 链符（展示用）
|
||||||
|
pct0 float 数据日涨幅（百分数，3.2 表示 +3.2%）
|
||||||
|
covered bool 有券商覆盖；upside float 预期空间（比例，0.25 表示 +25%）
|
||||||
|
risk_name bool 证券简称命中 ST / *ST / 退市族
|
||||||
|
accum_state str 吸筹状态文案（决策系统 fund_flow.state）；accum_score float；
|
||||||
|
accum_age int 评分日龄（交易日）
|
||||||
|
y_signal str 决策系统昨夜 signal_type
|
||||||
|
stale_snapshot bool 传导快照日与计划日不符
|
||||||
|
返回：verdict / reasons / missing / risk / gates / confirm。
|
||||||
|
"""
|
||||||
|
pct0 = _num(ev.get("pct0"))
|
||||||
|
upside = _num(ev.get("upside"))
|
||||||
|
accum_score = _num(ev.get("accum_score"))
|
||||||
|
accum_age = ev.get("accum_age")
|
||||||
|
accum_state = str(ev.get("accum_state") or "")
|
||||||
|
y_signal = str(ev.get("y_signal") or "").strip().upper()
|
||||||
|
|
||||||
|
gates = {
|
||||||
|
"pointed": bool(ev.get("pointed")),
|
||||||
|
"started": pct0 is not None and pct0 >= float(start_pct),
|
||||||
|
"covered": bool(ev.get("covered")) and upside is not None
|
||||||
|
and upside >= -float(neg_tol or 0.0),
|
||||||
|
"clean_name": not bool(ev.get("risk_name")),
|
||||||
|
}
|
||||||
|
accum_fresh = (isinstance(accum_age, int) and 0 <= accum_age <= int(accum_max_age))
|
||||||
|
confirm = accum_state.startswith("明确") and accum_fresh
|
||||||
|
|
||||||
|
risk: list[str] = []
|
||||||
|
if y_signal in BAD_SIGNALS:
|
||||||
|
risk.append(f"决策系统昨夜信号 {y_signal}")
|
||||||
|
if ev.get("stale_snapshot"):
|
||||||
|
risk.append("传导快照日与计划日不符")
|
||||||
|
if "派发" in accum_state:
|
||||||
|
risk.append("高位派发")
|
||||||
|
|
||||||
|
missing: list[str] = []
|
||||||
|
if pct0 is None:
|
||||||
|
missing.append("无行情")
|
||||||
|
if not ev.get("covered"):
|
||||||
|
missing.append("无券商覆盖")
|
||||||
|
if not accum_state:
|
||||||
|
missing.append("无吸筹评分（未入池）")
|
||||||
|
elif accum_state.startswith("明确") and not accum_fresh:
|
||||||
|
missing.append(f"吸筹评分陈旧 {accum_age} 日" if isinstance(accum_age, int)
|
||||||
|
else "吸筹评分日龄未知")
|
||||||
|
if not gates["pointed"]:
|
||||||
|
missing.append("无传导")
|
||||||
|
|
||||||
|
reasons: list[str] = []
|
||||||
|
if gates["pointed"]:
|
||||||
|
theme = ev.get("theme") or "环节"
|
||||||
|
n = ev.get("n_sources")
|
||||||
|
reasons.append(f"所在环节「{theme}」被{n}路指向" if n else f"所在环节「{theme}」被指向")
|
||||||
|
if gates["started"]:
|
||||||
|
reasons.append(f"当日已启动 {pct0:+.1f}%")
|
||||||
|
if confirm:
|
||||||
|
reasons.append(f"明确吸筹（评分 {accum_score:.0f}，{accum_age} 日前）"
|
||||||
|
if accum_score is not None else f"明确吸筹（{accum_age} 日前）")
|
||||||
|
if gates["covered"]:
|
||||||
|
reasons.append(f"券商覆盖，预期空间 {upside:+.0%}")
|
||||||
|
|
||||||
|
all_gates = all(gates.values())
|
||||||
|
only_missing_coverage = (gates["pointed"] and gates["started"] and gates["clean_name"]
|
||||||
|
and not gates["covered"])
|
||||||
|
if risk:
|
||||||
|
verdict = VERDICT_SHOW
|
||||||
|
elif all_gates and confirm:
|
||||||
|
verdict = VERDICT_CANDIDATE
|
||||||
|
elif only_missing_coverage or (all_gates and not confirm):
|
||||||
|
verdict = VERDICT_WATCH
|
||||||
|
else:
|
||||||
|
verdict = VERDICT_SHOW
|
||||||
|
|
||||||
|
failed = [k for k, ok in gates.items() if not ok]
|
||||||
|
return {"verdict": verdict, "reasons": reasons, "missing": missing, "risk": risk,
|
||||||
|
"gates": gates, "confirm": confirm, "failed_gates": failed}
|
||||||
|
|
||||||
|
|
||||||
|
def sort_key(row: dict[str, Any]) -> tuple:
|
||||||
|
"""卡内序：先判决（候选 < 关注 < 仅展示），再按数据日涨幅降序。不算总分。"""
|
||||||
|
v = _VERDICT_ORDER.get(row.get("verdict"), 9)
|
||||||
|
pct0 = _num(row.get("pct0"))
|
||||||
|
return (v, -(pct0 if pct0 is not None else -1e9))
|
||||||
5
pool.py
5
pool.py
|
|
@ -36,8 +36,9 @@ import db
|
||||||
import plan
|
import plan
|
||||||
|
|
||||||
|
|
||||||
# 决策系统结论里算「形态恶化」的定性（DROPPED 是它对掉出池子票的收尾标记）
|
# 决策系统结论里算「形态恶化」的定性（DROPPED 是它对掉出池子票的收尾标记）。
|
||||||
BAD_SIGNALS = {"SELL", "AVOID", "DROPPED"}
|
# 集合定义在 card.py（候选卡的硬风险用同一份），桥内只此一处；三处同源纪律见 card.py 模块说明。
|
||||||
|
from card import BAD_SIGNALS # noqa: E402,F401 —— 保留本名，调用方与单测不变
|
||||||
|
|
||||||
|
|
||||||
# ============================================================================
|
# ============================================================================
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,103 @@
|
||||||
|
-- astock-kg 候选卡插槽视图：三张只读视图，供 akg-factor-bridge 的候选卡与复盘消费。
|
||||||
|
-- 应用【桥机 155 · ~/project/akg-factor-bridge】：
|
||||||
|
-- docker compose exec -T akg-factor-bridge python run.py apply-views --file sql/astock_kg_card_views.sql --dry-run
|
||||||
|
-- docker compose exec -T -e AKG_PG_USER=<owner> -e AKG_PG_PASSWORD=<pw> \
|
||||||
|
-- akg-factor-bridge python run.py apply-views --file sql/astock_kg_card_views.sql
|
||||||
|
--
|
||||||
|
-- 与 sql/astock_kg_slot_views.sql 分开放：那份文件用 DROP + CREATE 且含 ALTER TABLE，
|
||||||
|
-- 整文件重跑会把生产在用的四张视图重删重建；本文件只用 CREATE OR REPLACE，反复应用无害。
|
||||||
|
-- 基座零代码：只读现有表 transmission_candidates、segment_members_projection、
|
||||||
|
-- segment_member_review、mkt_daily。应用时避开 06:10 的投影重建窗口。
|
||||||
|
-- 出处：docs/主观选股改进方案_2026-09-02.md 第 2.2 节第二项。
|
||||||
|
|
||||||
|
|
||||||
|
-- 一、已动成员视图 v_factor_transmission_moved
|
||||||
|
-- 被传导指向的环节（只认 Segment 目标；Concept 与 Company 目标的成员不在投影表里）的
|
||||||
|
-- 投影成员，减去传导台账里的未动名单 quiet，再减去复核账本判定"不属于本环节"的成员
|
||||||
|
-- （verdict = 'exclude'，与基座 transmission.scan 在算已动之前的剔除口径一致）。
|
||||||
|
-- 连接键用投影表的原名 COALESCE(orig_segment_name, segment_name)：投影期有改派重排，
|
||||||
|
-- 传导扫描用的目标名是图上原名。
|
||||||
|
-- 本视图只做集合差，不重算涨幅；"已动"的口径与基座 hotspot._movers_set 同源
|
||||||
|
-- （数据日涨幅达到 3%），涨幅本身从 v_factor_stock_daily 取。
|
||||||
|
-- 对账：按 (scan_date, target) 数本视图行数 + quiet 行数，与 transmission_candidates 的
|
||||||
|
-- members_total 比；不等的环节多半是基座扫描时成员集有上限（cap=200）而投影表没有，
|
||||||
|
-- 差异分"截断"与"真差异"两类，只对后者要求为零。
|
||||||
|
CREATE OR REPLACE VIEW v_factor_transmission_moved AS
|
||||||
|
WITH tc AS (
|
||||||
|
SELECT t.scan_date, t.rank, t.target, t.mkt_trade_date, t.chain_fit,
|
||||||
|
t.members_total, t.moved, t.moved_ratio,
|
||||||
|
(SELECT count(DISTINCT p->>'source')
|
||||||
|
FROM jsonb_array_elements(COALESCE(t.paths, '[]'::jsonb)) p) AS n_sources,
|
||||||
|
t.quiet
|
||||||
|
FROM transmission_candidates t
|
||||||
|
WHERE t.target_type = 'Segment'
|
||||||
|
),
|
||||||
|
quiet AS (
|
||||||
|
SELECT c.scan_date, c.target, q->>'ts_code' AS ts_code
|
||||||
|
FROM tc c
|
||||||
|
CROSS JOIN LATERAL jsonb_array_elements(COALESCE(c.quiet, '[]'::jsonb)) q
|
||||||
|
WHERE q->>'ts_code' IS NOT NULL AND q->>'ts_code' <> ''
|
||||||
|
),
|
||||||
|
members AS (
|
||||||
|
SELECT DISTINCT COALESCE(orig_segment_name, segment_name) AS seg,
|
||||||
|
ts_code, member_name, segment_name AS projected_segment
|
||||||
|
FROM segment_members_projection
|
||||||
|
WHERE ts_code IS NOT NULL AND ts_code <> ''
|
||||||
|
)
|
||||||
|
SELECT c.scan_date, c.target, c.rank, c.n_sources, c.chain_fit,
|
||||||
|
c.members_total, c.moved, c.moved_ratio, c.mkt_trade_date,
|
||||||
|
m.ts_code, m.member_name, m.projected_segment
|
||||||
|
FROM tc c
|
||||||
|
JOIN members m ON m.seg = c.target
|
||||||
|
LEFT JOIN quiet q
|
||||||
|
ON q.scan_date = c.scan_date AND q.target = c.target AND q.ts_code = m.ts_code
|
||||||
|
WHERE q.ts_code IS NULL
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM segment_member_review r
|
||||||
|
WHERE r.ts_code = m.ts_code AND r.segment_name = c.target AND r.verdict = 'exclude');
|
||||||
|
|
||||||
|
|
||||||
|
-- 二、个股日行情视图 v_factor_stock_daily
|
||||||
|
-- 候选卡"已启动"门槛（pct_change）与两条展示线（net_z 主力净额对自身二十日基线的
|
||||||
|
-- 标准化偏离、heat_chg 热度较往前第五个交易日的变化）的数据来源，只留行情列。
|
||||||
|
-- 吸筹不从这里取：基座落库的吸筹状态没有评分日，候选卡的吸筹三项只认决策系统结论表。
|
||||||
|
CREATE OR REPLACE VIEW v_factor_stock_daily AS
|
||||||
|
SELECT trade_date,
|
||||||
|
code AS ts_code,
|
||||||
|
(metrics->>'pct_change')::numeric AS pct_change,
|
||||||
|
(metrics->>'net_z')::numeric AS net_z,
|
||||||
|
(metrics->>'heat_chg')::numeric AS heat_chg,
|
||||||
|
(metrics->>'heat')::numeric AS heat,
|
||||||
|
(metrics->>'net_amount')::numeric AS net_amount
|
||||||
|
FROM mkt_daily
|
||||||
|
WHERE kind = 'stock';
|
||||||
|
|
||||||
|
|
||||||
|
-- 三、环节日行情视图 v_factor_segment_daily
|
||||||
|
-- 按投影表把个股日行情聚成环节行，只给原始值与成员明细集合；"主力净额前两名"、
|
||||||
|
-- "涨停近似（涨幅超 9.8%）"这类派生判断在桥的 card.py 算，不写进基座视图。
|
||||||
|
-- 投影表是当前态（每晨 06:10 删后插），历史日的成员集合带成员前视，复盘按冻结快照重建。
|
||||||
|
CREATE OR REPLACE VIEW v_factor_segment_daily AS
|
||||||
|
WITH members AS (
|
||||||
|
SELECT DISTINCT COALESCE(orig_segment_name, segment_name) AS seg, ts_code, member_name
|
||||||
|
FROM segment_members_projection
|
||||||
|
WHERE ts_code IS NOT NULL AND ts_code <> ''
|
||||||
|
)
|
||||||
|
SELECT s.trade_date,
|
||||||
|
p.seg AS segment_name,
|
||||||
|
count(*) AS members,
|
||||||
|
count(*) FILTER (WHERE s.pct_change >= 3) AS moved,
|
||||||
|
percentile_cont(0.5) WITHIN GROUP (ORDER BY s.pct_change) AS pct_median,
|
||||||
|
percentile_cont(0.5) WITHIN GROUP (ORDER BY s.heat_chg) AS heat_chg_median,
|
||||||
|
jsonb_agg(jsonb_build_object('ts_code', p.ts_code, 'name', p.member_name,
|
||||||
|
'pct', s.pct_change, 'net_amount', s.net_amount)
|
||||||
|
ORDER BY s.pct_change DESC NULLS LAST) AS members_detail
|
||||||
|
FROM members p
|
||||||
|
JOIN (
|
||||||
|
SELECT trade_date, code AS ts_code,
|
||||||
|
(metrics->>'pct_change')::numeric AS pct_change,
|
||||||
|
(metrics->>'heat_chg')::numeric AS heat_chg,
|
||||||
|
(metrics->>'net_amount')::numeric AS net_amount
|
||||||
|
FROM mkt_daily WHERE kind = 'stock'
|
||||||
|
) s ON s.ts_code = p.ts_code
|
||||||
|
GROUP BY s.trade_date, p.seg;
|
||||||
|
|
@ -0,0 +1,77 @@
|
||||||
|
"""card.judge() 纯逻辑单测（无需 DB、无需 pandas）。
|
||||||
|
|
||||||
|
覆盖：候选 / 关注（缺覆盖）/ 关注（缺确认）/ 仅展示（未启动）/ 仅展示（硬风险否决）
|
||||||
|
/ 缺失文案 / 评分陈旧 / 旋钮 / 卡内序 / 坏信号集合与 pool 同源。
|
||||||
|
|
||||||
|
跑法：python3 test_card.py 或 pytest test_card.py
|
||||||
|
"""
|
||||||
|
import card
|
||||||
|
|
||||||
|
BASE = dict(pointed=True, theme="散热器件", n_sources=2, pct0=4.1,
|
||||||
|
covered=True, upside=0.25, risk_name=False,
|
||||||
|
accum_state="明确吸筹·量在价先", accum_score=78, accum_age=3,
|
||||||
|
y_signal="WATCH", stale_snapshot=False)
|
||||||
|
|
||||||
|
|
||||||
|
def t(name, cond):
|
||||||
|
assert cond, name
|
||||||
|
print(" ok", name)
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
r = card.judge(dict(BASE))
|
||||||
|
t("三门槛全过且明确吸筹 -> 候选", r["verdict"] == "候选" and r["confirm"] and not r["risk"])
|
||||||
|
t("候选理由四条齐", len(r["reasons"]) == 4 and "已启动" in r["reasons"][1])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "covered": False, "upside": None})
|
||||||
|
t("只差覆盖 -> 关注", r["verdict"] == "关注" and "无券商覆盖" in r["missing"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "accum_state": "潜在吸筹·低位企稳"})
|
||||||
|
t("门槛全过但潜在吸筹不算确认 -> 关注", r["verdict"] == "关注" and not r["confirm"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "accum_state": ""})
|
||||||
|
t("无评分 -> 关注且缺失文案", r["verdict"] == "关注" and "无吸筹评分（未入池）" in r["missing"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "pct0": 1.2})
|
||||||
|
t("未启动 -> 仅展示", r["verdict"] == "仅展示" and "started" in r["failed_gates"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "pointed": False})
|
||||||
|
t("未被指向 -> 仅展示且标无传导", r["verdict"] == "仅展示" and "无传导" in r["missing"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "y_signal": "sell"})
|
||||||
|
t("决策系统 SELL -> 硬风险否决为仅展示", r["verdict"] == "仅展示" and r["risk"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "stale_snapshot": True})
|
||||||
|
t("快照日不符 -> 硬风险", r["verdict"] == "仅展示" and "传导快照日与计划日不符" in r["risk"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "accum_state": "⚠️ 高位派发"})
|
||||||
|
t("高位派发 -> 硬风险", r["verdict"] == "仅展示" and "高位派发" in r["risk"])
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "accum_age": 45})
|
||||||
|
t("评分陈旧 -> 不确认、关注、缺失文案", r["verdict"] == "关注" and not r["confirm"]
|
||||||
|
and any("陈旧 45" in m for m in r["missing"]))
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "pct0": 2.5}, start_pct=2.0)
|
||||||
|
t("启动阈值旋钮生效", r["verdict"] == "候选")
|
||||||
|
r = card.judge({**BASE, "accum_age": 45}, accum_max_age=60)
|
||||||
|
t("评分日龄旋钮生效", r["verdict"] == "候选")
|
||||||
|
r = card.judge({**BASE, "upside": -0.05}, neg_tol=0.10)
|
||||||
|
t("负容忍线传入生效", r["verdict"] == "候选")
|
||||||
|
r = card.judge({**BASE, "upside": -0.05})
|
||||||
|
t("负容忍默认 0：轻微为负不过覆盖门槛 -> 关注", r["verdict"] == "关注")
|
||||||
|
|
||||||
|
r = card.judge({**BASE, "risk_name": True})
|
||||||
|
t("ST 族 -> 仅展示", r["verdict"] == "仅展示" and "clean_name" in r["failed_gates"])
|
||||||
|
|
||||||
|
rows = [{"verdict": "关注", "pct0": 9.0}, {"verdict": "候选", "pct0": 3.5},
|
||||||
|
{"verdict": "候选", "pct0": 7.2}, {"verdict": "仅展示", "pct0": None}]
|
||||||
|
rows.sort(key=card.sort_key)
|
||||||
|
t("卡内序：候选在前、同判决涨幅降序、缺行情垫底",
|
||||||
|
[x["pct0"] for x in rows] == [7.2, 3.5, 9.0, None])
|
||||||
|
|
||||||
|
t("坏信号集合口径", card.BAD_SIGNALS == {"SELL", "AVOID", "DROPPED"})
|
||||||
|
print("ALL OK — 候选卡判决 / 关注两种 / 硬风险三种 / 缺失文案 / 旋钮 / 卡内序 全部通过")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
@ -0,0 +1,51 @@
|
||||||
|
"""代码版本号：解析挂载进容器的 .git 文件取短哈希，不需要 git 二进制。
|
||||||
|
|
||||||
|
## 为什么要有它
|
||||||
|
|
||||||
|
计划快照要带代码版本，复盘时才知道某天的候选单是哪一版规则算出来的。桥镜像是
|
||||||
|
python:3.11-slim、没装 git，freeze.py 里调 git 命令的取法在容器里恒为 unknown
|
||||||
|
（155 上 data/frozen/2026-09-01/manifest.json 实测）。仓库根挂在 /app，.git 目录随之
|
||||||
|
挂进来，直接读它就够了（2026-09-02 拍板：解析 .git 文件，不走宿主传环境变量）。
|
||||||
|
|
||||||
|
读法：.git/HEAD 是 "ref: refs/heads/<分支>" 就去读对应的 refs 文件；refs 文件不存在
|
||||||
|
（打包过）就在 .git/packed-refs 里找那一行；HEAD 本身就是裸哈希（游离头）直接用。
|
||||||
|
任何一步失败返回 "unknown"，绝不抛错——版本号缺失不该让计划断产。
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
def _read(p: Path) -> str | None:
|
||||||
|
try:
|
||||||
|
return p.read_text(encoding="utf-8").strip()
|
||||||
|
except OSError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def git_short_rev(repo_root: str | Path | None = None, length: int = 7) -> str:
|
||||||
|
roots = [Path(repo_root)] if repo_root else [Path("/app"), Path(__file__).resolve().parent]
|
||||||
|
for root in roots:
|
||||||
|
git = root / ".git"
|
||||||
|
head = _read(git / "HEAD")
|
||||||
|
if not head:
|
||||||
|
continue
|
||||||
|
if head.startswith("ref:"):
|
||||||
|
ref = head.split(":", 1)[1].strip()
|
||||||
|
sha = _read(git / ref)
|
||||||
|
if not sha:
|
||||||
|
packed = _read(git / "packed-refs") or ""
|
||||||
|
for line in packed.splitlines():
|
||||||
|
parts = line.split()
|
||||||
|
if len(parts) == 2 and parts[1] == ref:
|
||||||
|
sha = parts[0]
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
sha = head
|
||||||
|
if sha and len(sha) >= length and all(c in "0123456789abcdef" for c in sha[:length]):
|
||||||
|
return sha[:length]
|
||||||
|
return "unknown"
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
print(git_short_rev())
|
||||||
Loading…
Reference in New Issue