历史深度数据探查

README.md

@@ -41,8 +41,22 @@ akg-factor-bridge：读视图+热度 → 四路日截面变换 → 【漏斗合
 ## 用法（全程 Docker，不在宿主机直跑）
 
 ```bash
-# 0) 基座视图建一次/改一次——经 astock-kg 的 postgres 容器（容器名 akg-postgres，库/用户均 akg）
+# 0) 基座视图建一次/改一次。**按你在哪台机器上选一条**——桥是跨机部署的独立单元，
-docker exec -i akg-postgres psql -U akg -d akg < sql/astock_kg_slot_views.sql
+#    部署它的服务器上通常没有 astock-kg 的容器。
+#    (a) 在桥这边（推荐，无需 psql 客户端，用桥自己的 AKG_PG_* 连接）：
+docker compose exec -T akg-factor-bridge python run.py apply-views --dry-run   # 先看要跑哪几条
+docker compose exec -T akg-factor-bridge python run.py apply-views
+#        ⚠️ 需要 DDL 权限。日常只读账号会报 permission denied，用基座 owner 跑这一次：
+#        docker compose exec -T -e AKG_PG_USER=<owner> -e AKG_PG_PASSWORD=<pw> \
+#          akg-factor-bridge python run.py apply-views
+#        （必须用 -e 传进容器；写在 docker 前面只影响宿主机进程）
+#    (b) 在桥这边、想用原生 psql（一次性容器，不装客户端）：
+#        set -a; . ./.env; set +a
+#        docker run --rm -i -e PGPASSWORD="$AKG_PG_PASSWORD" postgres:16-alpine \
+#          psql -h "$AKG_PG_HOST" -p "${AKG_PG_PORT:-5432}" -U "$AKG_PG_USER" -d "$AKG_PG_DB" \
+#          -v ON_ERROR_STOP=1 < sql/astock_kg_slot_views.sql
+#    (c) 在 astock-kg 那台机器上（把本文件带过去）：
+#        docker exec -i akg-postgres psql -U akg -d akg < sql/astock_kg_slot_views.sql
 
 # 1) 配连接（地址填「桥容器可达」的：跨机=LAN IP；与基座同机同网=服务名，见下「网络」）
 cp .env.example .env && vim .env

apply_views.py

@@ -0,0 +1,124 @@
+"""把插槽视图 DDL 应用到基座 PG —— 用桥自己的 psycopg 连接，不依赖 psql 客户端。
+
+**为什么需要它**：桥是跨机部署的独立单元（设计 §4），部署它的服务器上通常
+**没有** astock-kg 的 `akg-postgres` 容器，README 里那条
+`docker exec -i akg-postgres psql ...` 只在基座那台机器上成立。
+而桥容器是 python:3.11-slim，也没有 psql 客户端。
+本模块用桥已有的 `AKG_PG_*` 连接执行 DDL，零新增凭据、全程 Docker。
+
+**边界说明**：视图定义文件本来就随桥仓库版本控制（sql/astock_kg_slot_views.sql），
+所以由桥来应用它并不越界——桥定义插槽接口，基座只是接口的宿主。
+但注意**权限**：日常运行只需只读账号，应用 DDL 需要能 CREATE VIEW / ALTER TABLE
+的账号（通常是基座 owner）。若报 permission denied，用 owner 账号跑一次即可：
+
+    docker compose exec -T \
+      -e AKG_PG_USER=<owner> -e AKG_PG_PASSWORD=<pw> \
+      akg-factor-bridge python run.py apply-views
+
+（注意必须用 `docker compose exec -e`：在命令前面写 `AKG_PG_USER=... docker compose exec`
+只会设置宿主机上 compose 客户端进程的环境变量，传不进容器。）
+
+逐语句执行 + 逐语句报错：一条失败（比如 ALTER 缺权限）不影响其余语句，
+这样你能准确看到是哪一条没过，而不是整个脚本回滚。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import psycopg
+
+import config
+
+DEFAULT_SQL = "sql/astock_kg_slot_views.sql"
+
+
+def split_statements(sql: str) -> list[str]:
+    """按分号切分 SQL 语句。剥掉 -- 行注释，尊重单引号字符串。
+
+    本文件的 DDL 里没有 $$ 引用块与多行注释，故不做处理；
+    若将来加了函数体，这里要先升级再用。
+    """
+    out, buf = [], []
+    in_str = in_comment = False
+    i, n = 0, len(sql)
+    while i < n:
+        c = sql[i]
+        nxt = sql[i + 1] if i + 1 < n else ""
+        if in_comment:
+            if c == "\n":
+                in_comment = False
+                buf.append(c)
+            i += 1
+            continue
+        if in_str:
+            buf.append(c)
+            if c == "'":
+                if nxt == "'":          # 转义的单引号 ''
+                    buf.append(nxt)
+                    i += 2
+                    continue
+                in_str = False
+            i += 1
+            continue
+        if c == "-" and nxt == "-":
+            in_comment = True
+            i += 2
+            continue
+        if c == "'":
+            in_str = True
+            buf.append(c)
+            i += 1
+            continue
+        if c == ";":
+            stmt = "".join(buf).strip()
+            if stmt:
+                out.append(stmt)
+            buf = []
+            i += 1
+            continue
+        buf.append(c)
+        i += 1
+    tail = "".join(buf).strip()
+    if tail:
+        out.append(tail)
+    return out
+
+
+def _label(stmt: str) -> str:
+    """给语句起个人能读的名字，用于逐条报告。"""
+    head = " ".join(stmt.split())[:78]
+    return head + ("…" if len(" ".join(stmt.split())) > 78 else "")
+
+
+def apply(path: str = DEFAULT_SQL, dry_run: bool = False) -> int:
+    p = Path(path)
+    if not p.exists():
+        p = Path("/app") / path          # 容器内工作目录兜底
+    if not p.exists():
+        raise SystemExit(f"找不到 SQL 文件: {path}")
+    stmts = split_statements(p.read_text(encoding="utf-8"))
+    c = config.akg_pg()
+    print(f"应用 {p}（{len(stmts)} 条语句）→ {c.user}@{c.host}:{c.port}/{c.db}")
+    if dry_run:
+        for i, s in enumerate(stmts, 1):
+            print(f"  [{i:>2}] {_label(s)}")
+        print("\n(dry-run，未执行)")
+        return 0
+
+    ok = fail = 0
+    # autocommit：逐条独立提交，一条失败不毒化后续（DDL 各自独立、且都幂等）
+    with psycopg.connect(host=c.host, port=c.port, user=c.user,
+                         password=c.password, dbname=c.db, autocommit=True) as conn:
+        for i, s in enumerate(stmts, 1):
+            try:
+                conn.execute(s)
+                ok += 1
+                print(f"  ✅ [{i:>2}] {_label(s)}")
+            except Exception as e:      # noqa: BLE001 —— 逐条报错才看得出是哪条缺权限
+                fail += 1
+                print(f"  ❌ [{i:>2}] {_label(s)}\n         {e!r}")
+    print(f"\n完成：成功 {ok} 条，失败 {fail} 条")
+    if fail:
+        print("提示：CREATE VIEW / ALTER TABLE 需要 DDL 权限；只读账号会报 "
+              "permission denied —— 用基座 owner 账号跑一次即可（见本文件 docstring）。")
+    return fail

docs/第一批验证清单_2026-07-26.md

@@ -42,11 +42,24 @@ git commit -m "fix: 交叉评审第一批——传导口径/事件回填/年报
 
 ## 1. 应用视图（基座侧，只建视图 + 加一个可空列 + 加索引）
 
+> ⚠️ **勘误（2026-07-26）**：原来这里写的 `docker exec -i akg-postgres psql ...`
+> 只在 **astock-kg 那台机器**上成立。桥是跨机部署的独立单元，桥的服务器上没有
+> `akg-postgres` 容器，桥容器本身（python:3.11-slim）也没有 psql 客户端。
+> 已新增 `run.py apply-views`，用桥自己的 `AKG_PG_*` 连接执行 DDL。
+
 ```bash
-docker exec -i akg-postgres psql -U akg -d akg < sql/astock_kg_slot_views.sql
+# 推荐：在桥这边跑，零新增凭据、全程 Docker
+docker compose exec -T akg-factor-bridge python run.py apply-views --dry-run   # 先看清单
+docker compose exec -T akg-factor-bridge python run.py apply-views
+
+# 若日常账号是只读的，会逐条报 permission denied —— 用基座 owner 跑这一次：
+docker compose exec -T \
+  -e AKG_PG_USER=<owner> -e AKG_PG_PASSWORD=<pw> \
+  akg-factor-bridge python run.py apply-views
+# 必须用 -e 传进容器：写在 docker 前面只设置宿主机进程的环境变量，进不去
 ```
 
-**预期**：`CREATE VIEW` ×4、`ALTER TABLE`、`CREATE INDEX`，无 ERROR。
+**预期**：6 条语句逐条 ✅（4 个 CREATE OR REPLACE VIEW + 1 个 ALTER TABLE + 1 个 CREATE INDEX）。
 **不动任何一行数据**，`claims` / `documents` 一个字节不变。
 
 ---

run.py

@@ -1,6 +1,7 @@
 """akg-factor-bridge CLI。
 
   python run.py views                                      # 连通性自检：打印视图/表行数
+  python run.py apply-views [--dry-run]                    # 把插槽视图 DDL 应用到基座 PG
   python run.py probe                                      # G1 体检（只读，详见 probe.py）
   python run.py freeze [--date D]                          # 输入冻结（G0.5，详见 freeze.py）
   python run.py register                                   # 注册四子因子到 factor_metadata
@@ -133,6 +134,9 @@ def main():
     p.add_argument("--section", choices=["all", "pools", "price", "upside", "corr"],
                    default="all",
                    help="pools=池结构清单 price=行情年表 upside=分布与q档位 corr=三项相关矩阵")
+    av = sub.add_parser("apply-views")
+    av.add_argument("--file", default="sql/astock_kg_slot_views.sql")
+    av.add_argument("--dry-run", action="store_true", help="只列语句不执行")
     f = sub.add_parser("freeze")
     f.add_argument("--date", help="默认今天")
     b = sub.add_parser("build")
@@ -149,6 +153,9 @@ def main():
     elif a.cmd == "probe":
         import probe  # 按需加载：一次性诊断命令，不影响常规链路
         probe.run(a.section)
+    elif a.cmd == "apply-views":
+        import apply_views
+        raise SystemExit(1 if apply_views.apply(a.file, a.dry_run) else 0)
     elif a.cmd == "freeze":
         import freeze
         freeze.snapshot(a.date)