#!/usr/bin/env python3
|
"""Fetch V2 evidence packages for uncovered industry-research securities.
|
|
The target list is derived at runtime from the four governed research pools and
|
the MySQL valuation ledger. Fetches run concurrently, while each V2 process
|
retains its own 90-second monotonic deadline and writes only to the analyst's
|
temporary workspace.
|
"""
|
|
from __future__ import annotations
|
|
import csv
|
import json
|
import os
|
import re
|
import subprocess
|
import sys
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from pathlib import Path
|
|
import pymysql
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
AS_OF = "2026-08-13"
|
OUT = ROOT / "ai-valuation-analyst" / "tmp" / "valuation_industry_research_gaps_20260814"
|
CACHE = ROOT / "ai-valuation-analyst" / "tmp" / "v2_cache"
|
|
|
def mysql_connection(database: str) -> pymysql.Connection:
|
return pymysql.connect(
|
host=os.environ.get("MYSQL_HOST", "127.0.0.1"),
|
port=int(os.environ.get("MYSQL_PORT", "3306")),
|
user=os.environ.get("MYSQL_USER", "root"),
|
password=os.environ["MYSQL_PASSWORD"],
|
database=database,
|
charset="utf8mb4",
|
autocommit=True,
|
cursorclass=pymysql.cursors.DictCursor,
|
)
|
|
|
def semiconductor_tickers() -> set[str]:
|
path = ROOT / "ana-data/cases/半导体案例/ANA-SEMI-20260722-001/outputs/核心文档/国内企业信息表.csv"
|
market_map = {"SSE": "SH", "SSE STAR": "SH", "SZSE": "SZ", "SZSE CHINEXT": "SZ", "BSE": "BJ"}
|
rows = csv.DictReader(path.open(encoding="utf-8-sig", newline=""))
|
result = set()
|
for row in rows:
|
code = row["ticker"].strip()
|
suffix = market_map.get(row["listed_market"].strip())
|
if suffix and re.fullmatch(r"\d{6}", code):
|
result.add(f"{code}.{suffix}")
|
# The North Exchange completed its 920-series code migration after the
|
# research snapshot. The local formal market contract uses the new code.
|
result.discard("835179.BJ")
|
result.add("920179.BJ")
|
return result
|
|
|
def robot_tickers() -> set[str]:
|
path = ROOT / "ana-data/cases/机器人案例/ANA-ROBOT-INDUSTRY-001/evidence/next_robot_036_market_database_company_universe_authority_20260726.csv"
|
return {row["symbol"].strip() for row in csv.DictReader(path.open(encoding="utf-8-sig", newline=""))}
|
|
|
def newenergy_tickers() -> set[str]:
|
path = ROOT / "ana-data/cases/新能源案例/核心文档/全量公司横向总表.md"
|
result = set()
|
for code in re.findall(r"\| (\d{6}) \| \[", path.read_text(encoding="utf-8")):
|
suffix = "SH" if code.startswith("6") else ("BJ" if code.startswith(("8", "9")) else "SZ")
|
result.add(f"{code}.{suffix}")
|
return result
|
|
|
def missing_tickers() -> list[str]:
|
universe = semiconductor_tickers() | robot_tickers() | newenergy_tickers()
|
with mysql_connection("stock_valuation") as connection, connection.cursor() as cursor:
|
cursor.execute("SELECT ticker FROM security")
|
covered = {row["ticker"] for row in cursor.fetchall()}
|
return sorted(universe - covered)
|
|
|
def fetch(ticker: str) -> dict[str, object]:
|
target = OUT / f"v2_{ticker.replace('.', '_')}"
|
env = os.environ.copy()
|
env["PYTHONPATH"] = str(ROOT / "dev/project-dev")
|
proc = subprocess.run(
|
[
|
sys.executable,
|
"-m",
|
"stock_valuation_pipeline_v2",
|
"--ticker",
|
ticker,
|
"--as-of",
|
AS_OF,
|
"--cache-dir",
|
str(CACHE),
|
"--output-dir",
|
str(target),
|
],
|
cwd=ROOT,
|
env=env,
|
capture_output=True,
|
text=True,
|
encoding="utf-8",
|
errors="replace",
|
timeout=120,
|
)
|
return {
|
"ticker": ticker,
|
"returncode": proc.returncode,
|
"stdout": proc.stdout.strip().splitlines()[-1] if proc.stdout.strip() else "",
|
"stderr": proc.stderr.strip()[-500:],
|
}
|
|
|
def latest_package_exists(ticker: str) -> bool:
|
return any(OUT.glob(f"v2_{ticker.replace('.', '_')}.failed-*")) or (OUT / f"v2_{ticker.replace('.', '_')}").exists()
|
|
|
def main() -> None:
|
if hasattr(sys.stdout, "reconfigure"):
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
tickers = missing_tickers()
|
if "--missing-packages-only" in sys.argv:
|
tickers = [ticker for ticker in tickers if not latest_package_exists(ticker)]
|
OUT.mkdir(parents=True, exist_ok=True)
|
rows: list[dict[str, object]] = []
|
with ThreadPoolExecutor(max_workers=12) as pool:
|
pending = {pool.submit(fetch, ticker): ticker for ticker in tickers}
|
for future in as_completed(pending):
|
ticker = pending[future]
|
try:
|
row = future.result()
|
except Exception as exc: # retain every batch failure
|
row = {"ticker": ticker, "returncode": 99, "error": repr(exc)}
|
rows.append(row)
|
print(f"[{len(rows):03d}/{len(tickers):03d}] {ticker} rc={row.get('returncode')}", flush=True)
|
rows.sort(key=lambda item: str(item["ticker"]))
|
manifest = {"as_of": AS_OF, "count": len(rows), "items": rows}
|
(OUT / "fetch_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
print(json.dumps({"count": len(rows), "returncodes": {str(code): sum(row.get("returncode") == code for row in rows) for code in sorted({int(row.get("returncode", 99)) for row in rows})}}, ensure_ascii=False))
|
|
|
if __name__ == "__main__":
|
main()
|
