akg-factor-bridge/config.py

"""连接配置与运行旋钮：全部从环境变量/.env 读取，不硬编码任何主机。

三处连接：
  AKG_PG_*       —— astock-kg 基座 PostgreSQL（读只读视图，只读账号即可）
  HEAT_MYSQL_*   —— 153 代理 MySQL（读 stock_fund_heat_scores 热度）
  FACTOR_MYSQL_* —— 平台 PROXY_DB_URL 指向的 MySQL（写 t_factor_* + factor_metadata）
现价来源 PRICE_MYSQL_*（upside 用）默认复用 FACTOR_MYSQL_* 同实例。
"""
import os
from dataclasses import dataclass

try:
    from dotenv import load_dotenv
    load_dotenv()
except Exception:
    pass


def _req(k: str) -> str:
    v = os.environ.get(k)
    if not v:
        raise RuntimeError(f".env 缺配置: {k}")
    return v


def _opt(k: str, fallback: str) -> str:
    return os.environ.get(k) or fallback


@dataclass(frozen=True)
class Conn:
    host: str
    port: int
    user: str
    password: str
    db: str


def akg_pg() -> Conn:
    return Conn(_req("AKG_PG_HOST"), int(os.environ.get("AKG_PG_PORT", 5432)),
                _req("AKG_PG_USER"), os.environ.get("AKG_PG_PASSWORD", ""), _req("AKG_PG_DB"))


def heat_mysql() -> Conn:
    return Conn(_req("HEAT_MYSQL_HOST"), int(os.environ.get("HEAT_MYSQL_PORT", 3306)),
                _req("HEAT_MYSQL_USER"), os.environ.get("HEAT_MYSQL_PASSWORD", ""), _req("HEAT_MYSQL_DB"))


def factor_mysql() -> Conn:
    return Conn(_req("FACTOR_MYSQL_HOST"), int(os.environ.get("FACTOR_MYSQL_PORT", 3306)),
                _req("FACTOR_MYSQL_USER"), os.environ.get("FACTOR_MYSQL_PASSWORD", ""), _req("FACTOR_MYSQL_DB"))


def price_mysql() -> Conn:
    """现价（gp_day_data）来源，默认与因子库同实例。"""
    return Conn(_opt("PRICE_MYSQL_HOST", _req("FACTOR_MYSQL_HOST")),
                int(_opt("PRICE_MYSQL_PORT", os.environ.get("FACTOR_MYSQL_PORT", "3306"))),
                _opt("PRICE_MYSQL_USER", _req("FACTOR_MYSQL_USER")),
                _opt("PRICE_MYSQL_PASSWORD", os.environ.get("FACTOR_MYSQL_PASSWORD", "")),
                _opt("PRICE_MYSQL_DB", _req("FACTOR_MYSQL_DB")))


def pms_mysql() -> Conn:
    """读持仓（首选 PMS 账本 pms_position，下游 trading_position 兜底）与决策系统
    结论（strategy_daily_results），入池用。这几张表都在 153 代理后面——与热度是
    同一台，默认直接复用 HEAT_MYSQL_*，只有代理路由不通时才需要单独配 PMS_MYSQL_*。"""
    return Conn(_opt("PMS_MYSQL_HOST", _req("HEAT_MYSQL_HOST")),
                int(_opt("PMS_MYSQL_PORT", os.environ.get("HEAT_MYSQL_PORT", "3306"))),
                _opt("PMS_MYSQL_USER", _req("HEAT_MYSQL_USER")),
                _opt("PMS_MYSQL_PASSWORD", os.environ.get("HEAT_MYSQL_PASSWORD", "")),
                _opt("PMS_MYSQL_DB", _req("HEAT_MYSQL_DB")))


@dataclass(frozen=True)
class MongoConn:
    host: str
    port: int
    user: str
    password: str
    db: str


def mongo() -> MongoConn:
    """选股计划入池的落点（决策系统每晚扫描读同一处）。"""
    return MongoConn(_req("MONGO_HOST"), int(os.environ.get("MONGO_PORT", 27017)),
                     _req("MONGO_USERNAME"), os.environ.get("MONGO_PASSWORD", ""),
                     _opt("MONGO_DB", "stock_predictions"))


# gp_day_data 股票代码列名（平台实测 = symbol；代码仍会自动兜底逐个试）
PRICE_CODE_COL = os.environ.get("PRICE_CODE_COL", "ts_code")
FACTOR_API_BASE = os.environ.get("FACTOR_API_BASE", "").rstrip("/")


# ============================================================================
# 运行旋钮（2026-07-26 评审引入）。全部有默认值，不填也能跑。
# ============================================================================

# --- 子因子是否受覆盖池限制：pool（现状）| market（评审 §4 建议）------------
# industry_pools 只存最新态、每周一 refresh_pools 自动长大。用它过滤**历史**子因子
# 会引入成员性前视（值点时正确、成员性前视——这是 §2.2 批传导用的同一条论证，
# 上升了一层），且截面统计量每周一结构性跳变。
# 子因子表的定位是「可独立观察的仪表」，过滤应发生在消费端（akg_score），不在这里。
# 传导不受此开关影响——它天生就是池内语义。
SUBFACTOR_UNIVERSE = os.environ.get("SUBFACTOR_UNIVERSE", "pool").strip().lower()

# --- 行情读取按月分块 -------------------------------------------------------
# 原来一次拉全区间，`--mode history --start 2006-01-01` 会把千万级行拉进 pandas。
PRICE_CHUNK_DAYS = int(os.environ.get("PRICE_CHUNK_DAYS", "31"))

# --- 因子表写入分块行数 -----------------------------------------------------
# 热度全史约 322×1674≈54 万行，原来单事务 executemany 走 ShardingSphere 代理有风险。
WRITE_CHUNK_ROWS = int(os.environ.get("WRITE_CHUNK_ROWS", "50000"))

# --- 事件来源白名单与单文档封顶（评审 §6.5：年报污染）----------------------
# documents.source_type 取值（astock-kg db/postgres/init.sql:12 已核实）：
#   annual_report / research_report / announcement / news / interactive_qa
# 年报也带 company_ts_code 文档锚，而一份年报能抽十几条 EVENT，且含**历史**诉讼/
# 处罚——会在年报披露日形成巨大负值尖峰。S2 若用 akg_event < −θ_e 做否决闸，
# 会因为一份年报提到历史诉讼而否决掉一只好股。默认只认公告。
EVENT_SOURCE_TYPES = {
    s.strip() for s in
    os.environ.get("EVENT_SOURCE_TYPES", "announcement").split(",") if s.strip()}
EVENT_MAX_PER_DOC = int(os.environ.get("EVENT_MAX_PER_DOC", "3"))

# --- 输入冻结落点（freeze.py）----------------------------------------------
# 容器内路径；docker-compose 已把仓库根挂到 /app，故默认落在仓库的 data/frozen/。
FROZEN_ROOT = os.environ.get("FROZEN_ROOT", "/app/data/frozen")

# --- 上游截断体检阈值（评审 硬伤1）-----------------------------------------
# 基座 graph_store.topic_context 的 Cypher 是 `LIMIT $cap`（默认 30）且无 ORDER BY；
# transmission.scan 落库时又有 `quiet[:12]`。命中这两个数就说明上游被截断了，
# 该行的 moved_ratio 建立在任意抽样上、覆盖也被展示逻辑锁住。
# 07-28 语义更新：基座二批后 topic_context cap=1000、scan 大主题闸=600、quiet 全量落库。
# 本值对齐基座"大主题闸"：members_total 超过它 = 基座制度回退，桥侧告警（factors.py）。
UPSTREAM_MEMBER_CAP = int(os.environ.get("UPSTREAM_MEMBER_CAP", "600"))

# --- 赛道门槛 C（G2：config/frontier_tracks.yml + tracks.py）-----------------
# yml 是唯一事实源（设计 §3.2）。C 闸默认关：映射还是 v0.1 草案，先跑
# `python run.py tracks` 做覆盖体检、清单转正后再置 1——届时主榜(gate=2)
# 再交赛道成员，观察档的产业链锚也并入赛道成员。
TRACKS_YML = os.environ.get("TRACKS_YML", "config/frontier_tracks.yml")
ENABLE_TRACK_GATE = os.environ.get("ENABLE_TRACK_GATE", "0") == "1"

# --- 重大风险股闸（07-31 拍板：默认杜绝）-------------------------------------
# ST / *ST / S(S)T / 退市族按证券简称识别，默认挡在档位之外——gate 记 0（不采纳），
# 因此不进主榜、观察档、计划与升降档。研究口径想看全貌时置 0 关闭。
EXCLUDE_RISK_NAMES = os.environ.get("EXCLUDE_RISK_NAMES", "1") == "1"

# --- 统一任务调度平台触发（08-03；键名与决策系统保持一致，便于平台侧统一配置）----
# 空串 = /api/v1/xxl/* 整组端点禁用（安全默认）。回调契约见 xxl.py 模块说明。
XXL_TRIGGER_KEY = os.environ.get("XXL_TRIGGER_KEY", "")
XXL_CALLBACK_READ_TIMEOUT = float(os.environ.get("XXL_CALLBACK_READ_TIMEOUT", "10"))
XXL_CALLBACK_MAX_RETRIES = int(os.environ.get("XXL_CALLBACK_MAX_RETRIES", "3"))

# --- 选股计划入池（08-03 定稿；规则与流程见 docs/选股计划入池_对接说明.md）------
# 写 Mongo stock_groups 的独立分组，决策系统每晚扫描按分组并集覆盖 → 候选票自动
# 获得夜间推理。入池范围与 PMS 候选同口径：强传导主榜前 POOL_TOP 只 + 当前持仓。
# 注意：POOL_TOP / POOL_TIERS 若调整，记得与 PMS 页面的 PMS_PLAN_TOP_N /
# PMS_PLAN_TIERS 保持一致，两边看到的候选才是同一批。
POOL_GROUP_CODE = os.environ.get("POOL_GROUP_CODE", "AKG_PLAN")
POOL_GROUP_NAME = os.environ.get("POOL_GROUP_NAME", "AKG每日选股计划池")
POOL_ORG_ID = os.environ.get("POOL_ORG_ID", "489281497140")
POOL_COLLECTION = os.environ.get("POOL_COLLECTION", "stock_groups")
POOL_RECYCLE_COLLECTION = os.environ.get("POOL_RECYCLE_COLLECTION", "stock_recycle_bin")
POOL_TOP = int(os.environ.get("POOL_TOP", "20"))
POOL_THEME_CAP = int(os.environ.get("POOL_THEME_CAP", "5"))
# 档位白名单（逗号分隔；空串=主榜全部）。默认只收强传导——候选宁缺毋滥。
POOL_TIERS = {s.strip() for s in os.environ.get("POOL_TIERS", "强传导").split(",") if s.strip()}
# 池子上限：计划+持仓+留池观察合计超过它时，从留池观察里清最久没上榜的。
# 上限直接决定决策系统每晚的推理量（每只全量分析约两三分钟）。
POOL_MAX = int(os.environ.get("POOL_MAX", "60"))
# 写完池子后触发决策系统的增量补扫（只补当天没分析过的票）。留空=不触发，
# 当晚 22:30 全量扫兜底。例：http://192.168.16.188:38000/api/v1/xxl/daily-scan
BIONIC_SCAN_URL = os.environ.get("BIONIC_SCAN_URL", "").rstrip("/")
BIONIC_SCAN_KEY = os.environ.get("BIONIC_SCAN_KEY", "")