#!/usr/bin/env python3
|
"""Evaluate current agriculture/new-energy research gaps and rebuild coverage.
|
|
This is a task coordinator, not a second valuation engine. It reuses the
|
registered V2 evidence collectors, the frozen V1 calculation core and the
|
accepted MySQL daily ledger. Agriculture and new-energy memberships are read
|
from their stable current research artifacts at runtime.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import importlib.util
|
import json
|
import os
|
import re
|
import subprocess
|
import sys
|
from collections import Counter
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from pathlib import Path
|
from typing import Any
|
|
import pymysql
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
BASE_SCRIPT = ROOT / "ai-valuation-analyst/tools/generate_industry_research_gap_valuations_20260814.py"
|
SPEC = importlib.util.spec_from_file_location("current_research_base", BASE_SCRIPT)
|
if SPEC is None or SPEC.loader is None:
|
raise RuntimeError(f"cannot import {BASE_SCRIPT}")
|
gen = importlib.util.module_from_spec(SPEC)
|
sys.modules[SPEC.name] = gen
|
SPEC.loader.exec_module(gen)
|
|
RESULT_ROOT = ROOT / "ana-data/result/股票估值"
|
SUMMARY_DIR = RESULT_ROOT / "行业调研标的估值覆盖"
|
V2_ROOT = ROOT / "ai-valuation-analyst/tmp/valuation_current_research_20260820"
|
V2_CACHE = ROOT / "ai-valuation-analyst/tmp/v2_cache"
|
BATCH_ID = "BATCH-STOCK-VALUATION-CURRENT-RESEARCH-20260820-001"
|
CASE_ROOT = ROOT / "ana-data/cases/股票估值" / BATCH_ID
|
|
|
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 write_json(path: Path, value: Any) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
def ticker_for(code: str) -> str:
|
if code.startswith("6"):
|
return f"{code}.SH"
|
if code.startswith(("0", "3")):
|
return f"{code}.SZ"
|
return f"{code}.BJ"
|
|
|
def research_targets() -> tuple[dict[str, Any], dict[str, str]]:
|
membership: dict[str, dict[str, Any]] = {}
|
source_notes: dict[str, str] = {}
|
|
def add(ticker: str, name: str, industry: str, layer: str) -> None:
|
item = membership.setdefault(ticker, {"names": [], "industries": set(), "layers": set()})
|
item["names"].append(name.strip())
|
item["industries"].add(industry)
|
item["layers"].add(layer)
|
|
fertilizer_path = ROOT / "ana-data/cases/农业案例/extracted/fertilizer_export_primary_disposition_2025.csv"
|
with fertilizer_path.open(encoding="utf-8-sig", newline="") as handle:
|
for row in csv.DictReader(handle):
|
add(
|
ticker_for(row["security_code"].strip()),
|
row["security_name_at_cutoff"].strip(),
|
"农业",
|
f"FERTILIZER:{row['primary_status'].strip()}",
|
)
|
source_notes["农业-化肥"] = f"{fertilizer_path.relative_to(ROOT).as_posix()}：固定50家公司"
|
|
selection_path = ROOT / "ana-data/cases/农业案例/extracted/candidate_selection.csv"
|
with selection_path.open(encoding="utf-8-sig", newline="") as handle:
|
for row in csv.DictReader(handle):
|
add(
|
ticker_for(row["security_code"].strip()),
|
row["security_name"].strip(),
|
"农业",
|
row["chain_code"].strip(),
|
)
|
source_notes["农业-首批"] = f"{selection_path.relative_to(ROOT).as_posix()}：农药/化肥16家公司"
|
|
company_root = ROOT / "ana-data/cases/新能源案例/核心文档/公司"
|
track_patterns = {
|
"SOLAR": r"光伏|太阳能|硅料|硅片|电池片|组件|逆变器",
|
"BATTERY": r"锂电|动力电池|电池材料",
|
"STORAGE": r"储能",
|
"WIND": r"风电|风机|风塔|叶片",
|
"NUCLEAR": r"核电|核能",
|
"POWER_SYSTEM": r"电力系统|电网|配电|输电",
|
}
|
card_count = 0
|
for path in sorted(company_root.glob("*.md")):
|
text = path.read_text(encoding="utf-8")
|
first_line = text.splitlines()[0] if text else ""
|
match = re.match(r"^#\s+(.+?)（(\d{6})）", first_line)
|
if not match:
|
continue
|
name, code = match.groups()
|
tracks = [track for track, pattern in track_patterns.items() if re.search(pattern, text)] or ["NEWENERGY_OTHER"]
|
for track in tracks:
|
add(ticker_for(code), name, "新能源", track)
|
card_count += 1
|
source_notes["新能源"] = f"{company_root.relative_to(ROOT).as_posix()}：{card_count}张稳定公司卡"
|
|
targets = {
|
ticker: gen.Target(
|
ticker=ticker,
|
research_name=next((name for name in item["names"] if name), ticker),
|
industries=tuple(sorted(item["industries"])),
|
research_layers=tuple(sorted(item["layers"])),
|
)
|
for ticker, item in membership.items()
|
}
|
return targets, source_notes
|
|
|
def existing_security() -> dict[str, str]:
|
with mysql_connection("stock_valuation") as connection, connection.cursor() as cursor:
|
cursor.execute("SELECT ticker,company FROM security WHERE active=1")
|
result = {row["ticker"]: row["company"] for row in cursor.fetchall()}
|
manifest_path = CASE_ROOT / "batch_manifest.json"
|
if manifest_path.exists():
|
prior = gen.base.read_json(manifest_path)
|
for row in prior.get("results", []):
|
if row.get("special_valuation_status"):
|
result[row["ticker"]] = row["company"]
|
return result
|
|
|
def package_dir(ticker: str) -> Path:
|
stem = f"v2_{ticker.replace('.', '_')}"
|
candidates = [path for path in V2_ROOT.glob(f"{stem}.failed-*") if (path / "provider_results.json").exists()]
|
exact = V2_ROOT / stem
|
if (exact / "provider_results.json").exists():
|
candidates.append(exact)
|
if not candidates:
|
raise FileNotFoundError(f"V2 package missing for {ticker}")
|
return max(candidates, key=lambda path: path.stat().st_mtime)
|
|
|
def load_provider(ticker: str) -> tuple[dict[str, Any], dict[str, Any], Path]:
|
directory = package_dir(ticker)
|
failure_path = directory / "failure.json"
|
failure = gen.base.read_json(failure_path) if failure_path.exists() else {"status": "COMPLETE"}
|
return gen.base.read_json(directory / "provider_results.json"), failure, directory
|
|
|
def fetch_one(ticker: str, as_of: str) -> dict[str, Any]:
|
try:
|
directory = package_dir(ticker)
|
return {"ticker": ticker, "returncode": 0, "status": "REUSED", "package": directory.name}
|
except FileNotFoundError:
|
pass
|
target = V2_ROOT / 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(V2_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,
|
"status": "FETCHED",
|
"stdout_tail": proc.stdout.strip().splitlines()[-1] if proc.stdout.strip() else "",
|
"stderr_tail": proc.stderr.strip()[-500:],
|
}
|
|
|
def fetch_missing(as_of: str, workers: int) -> dict[str, Any]:
|
targets, _ = research_targets()
|
covered = existing_security()
|
missing = sorted(set(targets) - set(covered))
|
V2_ROOT.mkdir(parents=True, exist_ok=True)
|
rows: list[dict[str, Any]] = []
|
with ThreadPoolExecutor(max_workers=workers) as pool:
|
pending = {pool.submit(fetch_one, ticker, as_of): ticker for ticker in missing}
|
for future in as_completed(pending):
|
ticker = pending[future]
|
try:
|
row = future.result()
|
except Exception as exc:
|
row = {"ticker": ticker, "returncode": 99, "status": "ERROR", "error": repr(exc)}
|
rows.append(row)
|
print(f"[{len(rows):03d}/{len(missing):03d}] {ticker} {row['status']} rc={row['returncode']}", flush=True)
|
rows.sort(key=lambda item: item["ticker"])
|
manifest = {"batch_id": BATCH_ID, "as_of": as_of, "missing_before": len(missing), "items": rows}
|
write_json(V2_ROOT / "fetch_manifest.json", manifest)
|
return manifest
|
|
|
PE_ADDITIONS = {
|
"agri_resource": ((6, 10), (10, 16), (16, 24)),
|
"agri_fertilizer": ((7, 11), (11, 18), (18, 26)),
|
"agri_pesticide": ((10, 15), (15, 24), (24, 34)),
|
"agri_distribution": ((7, 11), (11, 17), (17, 24)),
|
}
|
|
|
def classify(target: Any, company: str) -> tuple[str, str, str, str]:
|
layers = " ".join(target.research_layers)
|
if "农业" in target.industries:
|
if "PESTICIDE" in layers or company in {"海利尔", "美邦股份"}:
|
return "agri_pesticide", "农药原药、制剂或农化服务", "产品价格、销量、登记证、渠道、原料成本和出口需求", "农药周期、环保安监、库存渠道、出口和价格波动"
|
if company in {"辉隆股份", "浙农股份", "天禾股份", "中农立华", "富邦科技", "农发种业"}:
|
return "agri_distribution", "农资流通、助剂或综合农业服务", "销量、周转、渠道效率、费用率和营运资金", "低毛利、应收存货、渠道信用和业务纯度"
|
if company in {"藏格矿业", "盐湖股份", "亚钾国际", "川发龙蟒", "川恒股份", "洛阳钼业", "甘肃能化", "兰花科创", "兖矿能源", "中煤能源"}:
|
return "agri_resource", "资源型化肥原料或跨业务资源品", "资源价格、产销量、现金成本、权益产能和资本开支", "商品周期、业务纯度、资源税费、扩产和峰值利润外推"
|
return "agri_fertilizer", "氮磷钾肥、复合肥或化工联产", "肥价、原料价差、产销量、开工率、出口和产品结构", "化肥周期、煤气磷硫成本、出口政策、库存和现金流"
|
if "新能源" in target.industries:
|
if any(track in layers for track in ("SOLAR", "BATTERY")):
|
return "newenergy_cycle", "光伏或锂电材料、设备、组件与系统", "产品价格、出货、产能利用率、单位成本和库存", "产能过剩、价格战、减值、技术迭代和现金流"
|
return "newenergy_stable", "储能、风电、核电或新型电力系统", "订单、装机、交付、利用小时、项目验收和回款", "项目周期、政策、电价、应收回款和资本开支"
|
return gen._original_classify(target, company)
|
|
|
def bubble_fields(price: float, base_high: float, optimistic_high: float, tier: str, market: dict[str, Any], business: str) -> dict[str, Any]:
|
if not tier.startswith("agri_"):
|
return gen._original_bubble_fields(price, base_high, optimistic_high, tier, market, business)
|
premium = price / base_high - 1 if base_high > 0 else None
|
vs_optimistic = price / optimistic_high - 1 if optimistic_high > 0 else None
|
if premium is None or premium <= 0:
|
return {
|
"status": "未识别估值泡沫",
|
"primary_cause": "当前价未超过基准合理区间上沿",
|
"secondary_cause": "不代表没有周期和经营风险",
|
"nature": "规则判定",
|
"confidence": "中",
|
"reason": f"当前价未超过基准上沿；{business}的价格、成本或现金流变化仍可能令合理区间下修。",
|
}
|
if vs_optimistic is not None and vs_optimistic > 0:
|
status = "泡沫-极端"
|
elif premium <= 0.15:
|
status = "泡沫-轻"
|
elif premium <= 0.50:
|
status = "泡沫-中"
|
else:
|
status = "泡沫-高"
|
causes = {
|
"agri_resource": "资源品稀缺性、价格上行或高景气持续预期",
|
"agri_fertilizer": "肥价修复、出口与成本价差改善预期",
|
"agri_pesticide": "农药周期修复、出口和库存去化预期",
|
"agri_distribution": "渠道整合、周转改善与农业服务成长预期",
|
}
|
cause = causes[tier]
|
secondary = "趋势动量与交易拥挤放大溢价" if (market.get("distance_to_ma60") or 0) > 0.10 else "乐观利润和估值倍数被提前定价"
|
return {
|
"status": status,
|
"primary_cause": cause,
|
"secondary_cause": secondary,
|
"nature": "基于估值反推、业务驱动和前复权量价的推断",
|
"confidence": "中低",
|
"reason": f"当前价高于基准上沿{premium * 100:.1f}%；市场最可能提前交易{cause}，{secondary}。周期利润未被按永久高位资本化。",
|
}
|
|
|
def bounded_financials(provider: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
|
"""Allow one disclosed cash-flow schema omission without inventing profit.
|
|
*ST雅博's registered cash-flow payload omits CONSTRUCT_LONG_ASSET. Capex is
|
therefore set to zero only for the free-cash-flow cross-check; the formal
|
report/evidence manifest records that FCF is an upper bound. Profit,
|
equity, debt and valuation inputs keep their ordinary parser contracts.
|
"""
|
company = (provider.get("announcements", {}).get("identity") or {}).get("company", "")
|
if company not in {"*ST雅博", "ST雅博"}:
|
return gen._original_financials(provider)
|
original_number = gen.base._number
|
|
def optional_capex(row: dict[str, Any], *keys: str) -> float:
|
if keys == ("CONSTRUCT_LONG_ASSET",) and not any(row.get(key) is not None for key in keys):
|
return 0.0
|
return original_number(row, *keys)
|
|
gen.base._number = optional_capex
|
try:
|
return gen._original_financials(provider)
|
finally:
|
gen.base._number = original_number
|
|
|
def disclose_bounded_capex(row: dict[str, Any]) -> None:
|
if row["ticker"] != "002323.SZ":
|
return
|
report_path = ROOT / row["formal_path"]
|
report_path.write_text(
|
report_path.read_text(encoding="utf-8")
|
+ "\n## 有界数据缺口\n\n"
|
+ "- 注册现金流负载未提供 `CONSTRUCT_LONG_ASSET`（购建长期资产现金支出）字段；本报告仅在现金流交叉验证中按0处理，因此自由现金流是上限，不作为抬高合理区间的主锚。\n"
|
+ "- 利润、归母净资产、股本和三情景估值未使用该缺失字段；后续取得法定现金流明细时复核自由现金流质量。\n",
|
encoding="utf-8",
|
)
|
evidence_path = report_path.parent / "source_evidence_manifest.json"
|
evidence = gen.base.read_json(evidence_path)
|
evidence.setdefault("bounded_gaps", []).append(
|
"*ST雅博现金流镜像缺CONSTRUCT_LONG_ASSET；capex=0仅用于FCF上限交叉验证，不进入主估值锚。"
|
)
|
write_json(evidence_path, evidence)
|
|
|
def configure_generator(as_of: str) -> None:
|
stamp = as_of.replace("-", "")
|
gen.AS_OF = as_of
|
gen.STAMP = stamp
|
gen.BATCH_ID = BATCH_ID
|
gen.RESULT_ROOT = RESULT_ROOT
|
gen.SUMMARY_DIR = SUMMARY_DIR
|
gen.CASE_ROOT = CASE_ROOT
|
gen.V2_ROOT = V2_ROOT
|
gen.CACHE = V2_CACHE / "blobs/sha256"
|
gen.base.AS_OF = as_of
|
gen.base.RESULT_ROOT = RESULT_ROOT
|
gen.base.CASE_ROOT = CASE_ROOT
|
gen.base.CACHE = gen.CACHE
|
gen.PE_MULTIPLES.update(PE_ADDITIONS)
|
if not hasattr(gen, "_original_classify"):
|
gen._original_classify = gen.classify
|
if not hasattr(gen, "_original_bubble_fields"):
|
gen._original_bubble_fields = gen.bubble_fields
|
if not hasattr(gen, "_original_financials"):
|
gen._original_financials = gen.financials
|
gen.classify = classify
|
gen.bubble_fields = bubble_fields
|
gen.financials = bounded_financials
|
gen.failed_dir = package_dir
|
gen.load_provider = load_provider
|
|
|
def generate_missing(as_of: str) -> dict[str, Any]:
|
configure_generator(as_of)
|
targets, source_notes = research_targets()
|
before = existing_security()
|
missing = [targets[ticker] for ticker in sorted(set(targets) - set(before))]
|
prior_manifest_path = CASE_ROOT / "batch_manifest.json"
|
prior_manifest = gen.base.read_json(prior_manifest_path) if prior_manifest_path.exists() else {}
|
prior_results = {row["ticker"]: row for row in prior_manifest.get("results", [])}
|
rows: list[dict[str, Any]] = []
|
failures: list[dict[str, str]] = []
|
for index, target in enumerate(missing, 1):
|
try:
|
row = gen.build(target)
|
disclose_bounded_capex(row)
|
rows.append(row)
|
print(f"[{index:03d}/{len(missing):03d}] OK {target.ticker} {row['company']}", flush=True)
|
except Exception as exc:
|
error = f"{type(exc).__name__}: {exc}"
|
failures.append({"ticker": target.ticker, "company": target.research_name, "error": error})
|
print(f"[{index:03d}/{len(missing):03d}] FAIL {target.ticker} {error}", flush=True)
|
|
CASE_ROOT.mkdir(parents=True, exist_ok=True)
|
all_results = dict(prior_results)
|
all_results.update({row["ticker"]: row for row in rows})
|
result_rows = [all_results[ticker] for ticker in sorted(all_results)]
|
manifest = {
|
"batch_id": BATCH_ID,
|
"as_of": as_of,
|
"research_target_count": len(targets),
|
"existing_before": prior_manifest.get("existing_before", len(set(targets) & set(before))),
|
"missing_before": prior_manifest.get("missing_before", len(missing)),
|
"success": len(result_rows),
|
"failures": failures,
|
"sources": source_notes,
|
"results": result_rows,
|
}
|
write_json(CASE_ROOT / "batch_manifest.json", manifest)
|
|
SUMMARY_DIR.mkdir(parents=True, exist_ok=True)
|
detail_path = SUMMARY_DIR / "农业新能源新增估值明细.csv"
|
columns = [
|
"ticker", "company", "research_name", "industries", "research_layers", "price_date", "price",
|
"normalized_profit", "normalized_pe", "pb", "consensus_count", "consensus_2026", "forward_pe",
|
"base_low", "base_high", "optimistic_high", "label", "bubble_status", "bubble_primary_cause",
|
"bubble_reason", "special_valuation_status", "qa", "share_status", "formal_path",
|
]
|
with detail_path.open("w", encoding="utf-8-sig", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=columns, extrasaction="ignore")
|
writer.writeheader()
|
for row in result_rows:
|
item = dict(row)
|
item["industries"] = ";".join(row["industries"])
|
item["research_layers"] = ";".join(row["research_layers"])
|
item["qa"] = row["qa"]["status"]
|
writer.writerow(item)
|
labels = Counter(row["label"] for row in result_rows)
|
lines = [
|
"# 农业与新能源研究标的新增基础估值",
|
"",
|
f"- 估值日：{as_of}；初始范围缺口：{manifest['missing_before']}只；已形成结论：{len(result_rows)}只；当前失败：{len(failures)}只。",
|
f"- 标签：偏低{labels['偏低']}、基本合理{labels['基本合理']}、偏贵{labels['偏贵']}、明显偏贵{labels['明显偏贵']}。",
|
"- 估值区间是条件化研究结论，不是目标价、交易指令或收益承诺。",
|
"",
|
"| 公司 | 代码 | 行业 | 收盘价 | 基准合理区间 | 判断 | 泡沫 |",
|
"|---|---|---|---:|---:|---|---|",
|
]
|
for row in result_rows:
|
link = "../../../../" + row["formal_path"]
|
lines.append(
|
f"| [{row['company']}]({link}) | {row['ticker']} | {'、'.join(row['industries'])} | {row['price']:.2f}元 | "
|
f"{row['base_low']:.2f}—{row['base_high']:.2f}元 | {row['label']} | {row['bubble_status']} |"
|
)
|
if failures:
|
lines += ["", "## 明确失败", ""] + [f"- {item['ticker']} {item['company']}：{item['error']}" for item in failures]
|
lines += ["", f"机器明细：`{detail_path.name}`。", ""]
|
(SUMMARY_DIR / "农业与新能源研究标的新增估值汇总.md").write_text("\n".join(lines), encoding="utf-8")
|
return manifest
|
|
|
def merged_research_membership() -> dict[str, dict[str, Any]]:
|
merged: dict[str, dict[str, Any]] = {}
|
|
def add(ticker: str, name: str, industries: list[str], layers: list[str]) -> None:
|
item = merged.setdefault(ticker, {"name": name, "industries": set(), "layers": set()})
|
if name and not item["name"]:
|
item["name"] = name
|
item["industries"].update(value for value in industries if value)
|
item["layers"].update(value for value in layers if value)
|
|
old_path = SUMMARY_DIR / "四行业调研标的估值覆盖清单.csv"
|
with old_path.open(encoding="utf-8-sig", newline="") as handle:
|
for row in csv.DictReader(handle):
|
add(row["ticker"], row.get("research_name") or row.get("company") or "", row.get("industries", "").split(";"), row.get("research_layers", "").split(";"))
|
targets, _ = research_targets()
|
for ticker, target in targets.items():
|
add(ticker, target.research_name, list(target.industries), list(target.research_layers))
|
return merged
|
|
|
def rebuild_coverage(trade_date: str) -> dict[str, Any]:
|
membership = merged_research_membership()
|
with mysql_connection("stock_valuation") as connection, connection.cursor() as cursor:
|
cursor.execute("SELECT ticker,company FROM security WHERE active=1")
|
securities = {row["ticker"]: row["company"] for row in cursor.fetchall()}
|
cursor.execute(
|
"SELECT j.ticker,s.company,j.trade_date,j.close,j.base_low,j.base_high,j.optimistic_high,j.label,"
|
"v.valuation_date,v.report_path FROM daily_judgement j JOIN security s ON s.ticker=j.ticker "
|
"JOIN valuation_version v ON v.valuation_id=j.valuation_id WHERE j.trade_date=%s",
|
(trade_date,),
|
)
|
judgements = {row["ticker"]: row for row in cursor.fetchall()}
|
|
rows: list[dict[str, Any]] = []
|
for ticker, item in sorted(membership.items()):
|
current = judgements.get(ticker, {})
|
if current:
|
status = "已按最新交易日判定"
|
elif ticker in securities:
|
status = "已有基础估值但当日无有效价格"
|
else:
|
status = "尚无数值基础估值或属于特殊结论"
|
rows.append({
|
"ticker": ticker,
|
"company": current.get("company") or securities.get(ticker) or item["name"],
|
"industries": ";".join(sorted(item["industries"])),
|
"research_layers": ";".join(sorted(item["layers"])),
|
"coverage_status": status,
|
"trade_date": str(current.get("trade_date") or ""),
|
"close": current.get("close", ""),
|
"base_low": current.get("base_low", ""),
|
"base_high": current.get("base_high", ""),
|
"optimistic_high": current.get("optimistic_high", ""),
|
"label": current.get("label", ""),
|
"valuation_date": str(current.get("valuation_date") or ""),
|
"report_path": current.get("report_path", ""),
|
})
|
|
SUMMARY_DIR.mkdir(parents=True, exist_ok=True)
|
csv_path = SUMMARY_DIR / "全部行业调研标的估值覆盖清单.csv"
|
with csv_path.open("w", encoding="utf-8-sig", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
|
writer.writeheader()
|
writer.writerows(rows)
|
|
industries = sorted({industry for item in membership.values() for industry in item["industries"]})
|
lines = [
|
"# 全部行业调研标的估值覆盖",
|
"",
|
f"- 当前价格日：{trade_date}；研究证券去重后{len(rows)}只。",
|
"- 证券池来自半导体、机器人、军工既有正式覆盖清单，以及农业、新能源当前稳定研究成果；全A股底表和待核验发现候选不纳入。",
|
"- 当前价格判定来自 MySQL `stock_valuation`；合理价值区间为条件化估值，不是目标价或交易指令。",
|
"",
|
"## 覆盖统计",
|
"",
|
"| 行业 | 调研证券 | 最新交易日已判定 | 已有估值但缺当日价 | 尚无数值估值/特殊结论 |",
|
"|---|---:|---:|---:|---:|",
|
]
|
for industry in industries:
|
pool = [row for row in rows if industry in row["industries"].split(";")]
|
counts = Counter(row["coverage_status"] for row in pool)
|
lines.append(
|
f"| {industry} | {len(pool)} | {counts['已按最新交易日判定']} | "
|
f"{counts['已有基础估值但当日无有效价格']} | {counts['尚无数值基础估值或属于特殊结论']} |"
|
)
|
labels = Counter(row["label"] for row in rows if row["label"])
|
lines += [
|
"",
|
"## 最新价格判断分布（研究池去重）",
|
"",
|
f"- 偏低{labels['偏低']}只、基本合理{labels['基本合理']}只、偏贵{labels['偏贵']}只、明显偏贵{labels['明显偏贵']}只。",
|
f"- 逐股完整清单：[{csv_path.name}]({csv_path.name})。",
|
"- 半导体179只按估值由便宜到贵的完整结果：[Markdown](半导体估值排序.md)；[CSV](半导体估值排序.csv)。",
|
"- 全部已评估证券（包括用户单独点名、但不属于上述行业研究池的公司）的唯一当前入口仍是 `../估值台账/latest.md` 与 `../估值台账/latest.csv`。",
|
"",
|
]
|
md_path = SUMMARY_DIR / "全部行业调研标的估值覆盖汇总.md"
|
md_path.write_text("\n".join(lines), encoding="utf-8")
|
|
focus_rows = [row for row in rows if set(row["industries"].split(";")) & {"农业", "新能源"}]
|
label_order = {"偏低": 0, "基本合理": 1, "偏贵": 2, "明显偏贵": 3, "": 4}
|
focus_rows.sort(key=lambda row: (label_order.get(row["label"], 5), row["ticker"]))
|
focus_lines = [
|
"# 农业与新能源研究标的当前估值列表",
|
"",
|
f"- 当前完整价格日：{trade_date}；农业56只，新能源115只，合计去重{len(focus_rows)}只。",
|
"- 顺序为偏低、基本合理、偏贵、明显偏贵、特殊/缺口；同档按证券代码排列。",
|
"- 合理区间是条件化估值，不是目标价、交易指令或收益承诺。",
|
"",
|
]
|
for industry in ("农业", "新能源"):
|
pool = [row for row in focus_rows if industry in row["industries"].split(";")]
|
counts = Counter(row["label"] or "特殊/缺口" for row in pool)
|
focus_lines.append(
|
f"- {industry}：偏低{counts['偏低']}、基本合理{counts['基本合理']}、偏贵{counts['偏贵']}、"
|
f"明显偏贵{counts['明显偏贵']}、特殊/缺口{counts['特殊/缺口']}。"
|
)
|
focus_lines += [
|
"",
|
"| 公司 | 代码 | 行业 | 价格日 | 收盘价 | 基准合理区间 | 判断 | 状态 | 正式报告 |",
|
"|---|---|---|---|---:|---:|---|---|---|",
|
]
|
for row in focus_rows:
|
close = "—" if row["close"] == "" else f"{float(row['close']):.2f}元"
|
base = "—" if row["base_low"] == "" else f"{float(row['base_low']):.2f}—{float(row['base_high']):.2f}元"
|
report = "—" if not row["report_path"] else f"[报告](../../../../{row['report_path']})"
|
focus_lines.append(
|
f"| {row['company']} | {row['ticker']} | {row['industries'].replace(';', '、')} | "
|
f"{row['trade_date'] or '—'} | {close} | {base} | {row['label'] or '特殊/缺口'} | "
|
f"{row['coverage_status']} | {report} |"
|
)
|
focus_path = SUMMARY_DIR / "农业与新能源研究标的当前估值列表.md"
|
focus_path.write_text("\n".join(focus_lines) + "\n", encoding="utf-8")
|
manifest = {
|
"batch_id": BATCH_ID,
|
"trade_date": trade_date,
|
"target_count": len(rows),
|
"current_judgement_count": sum(bool(row["trade_date"]) for row in rows),
|
"industry_counts": {industry: sum(industry in row["industries"].split(";") for row in rows) for industry in industries},
|
"label_counts": dict(labels),
|
"csv_sha256": hashlib.sha256(csv_path.read_bytes()).hexdigest(),
|
"md_sha256": hashlib.sha256(md_path.read_bytes()).hexdigest(),
|
"agriculture_newenergy_md_sha256": hashlib.sha256(focus_path.read_bytes()).hexdigest(),
|
}
|
ranking_md = SUMMARY_DIR / "半导体估值排序.md"
|
ranking_csv = SUMMARY_DIR / "半导体估值排序.csv"
|
if ranking_md.is_file() and ranking_csv.is_file():
|
manifest["semiconductor_ranking_md_sha256"] = hashlib.sha256(ranking_md.read_bytes()).hexdigest()
|
manifest["semiconductor_ranking_csv_sha256"] = hashlib.sha256(ranking_csv.read_bytes()).hexdigest()
|
write_json(SUMMARY_DIR / "全部行业调研标的估值覆盖manifest.json", manifest)
|
return manifest
|
|
|
def repair_generated_labels(as_of: str) -> dict[str, Any]:
|
"""Align the human conclusion with the frozen V1/ledger interval rule."""
|
configure_generator(as_of)
|
manifest_path = CASE_ROOT / "batch_manifest.json"
|
manifest = gen.base.read_json(manifest_path)
|
changed: list[dict[str, str]] = []
|
for row in manifest.get("results", []):
|
expected = gen.label_for(
|
float(row["price"]),
|
(float(row["base_low"]), float(row["base_high"])),
|
(float(row["base_high"]), float(row["optimistic_high"])),
|
)
|
old = row["label"]
|
if expected == old:
|
continue
|
report_path = ROOT / row["formal_path"]
|
text = report_path.read_text(encoding="utf-8")
|
text = text.replace(f"，判断为{old}；", f"，判断为{expected}；", 1)
|
report_path.write_text(text, encoding="utf-8")
|
evidence_path = report_path.parent / "source_evidence_manifest.json"
|
evidence = gen.base.read_json(evidence_path)
|
evidence["checks"]["label"] = expected
|
write_json(evidence_path, evidence)
|
row["label"] = expected
|
changed.append({"ticker": row["ticker"], "old": old, "new": expected})
|
write_json(manifest_path, manifest)
|
generate_missing(as_of)
|
return {"changed_count": len(changed), "changed": changed}
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("action", choices=("inventory", "fetch", "generate", "coverage", "repair"))
|
parser.add_argument("--as-of", default="2026-08-19")
|
parser.add_argument("--workers", type=int, default=12)
|
args = parser.parse_args()
|
if hasattr(sys.stdout, "reconfigure"):
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
if args.action == "inventory":
|
targets, notes = research_targets()
|
covered = existing_security()
|
result = {"targets": len(targets), "covered": len(set(targets) & set(covered)), "missing": len(set(targets) - set(covered)), "sources": notes}
|
elif args.action == "fetch":
|
result = fetch_missing(args.as_of, args.workers)
|
elif args.action == "generate":
|
result = generate_missing(args.as_of)
|
elif args.action == "repair":
|
result = repair_generated_labels(args.as_of)
|
else:
|
result = rebuild_coverage(args.as_of)
|
print(json.dumps(result, ensure_ascii=False, default=str))
|
|
|
if __name__ == "__main__":
|
main()
|
