#!/usr/bin/env python3
|
"""Mid-year point-in-time refresh for every active stock valuation.
|
|
This is a valuation-analyst task utility, not a second production service. It
|
uses the accepted MySQL ledger as the universe, Eastmoney's registered
|
structured financial mirror for report facts, and dated institution details
|
plus business identity from 10jqka. Raw responses remain in the valuation
|
temporary area; formal outputs are written by the later revalue/publish phase.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import json
|
import re
|
import subprocess
|
import sys
|
import threading
|
import time
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Any
|
|
import requests
|
from bs4 import BeautifulSoup
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
BATCH_ID = "BATCH-STOCK-VALUATION-MIDYEAR-FULL-REFRESH-20260826-001"
|
TMP_ROOT = ROOT / "ana-data/tmp/股票估值" / BATCH_ID
|
RAW_ROOT = TMP_ROOT / "raw"
|
CASE_ROOT = ROOT / "ana-data/cases/股票估值" / BATCH_ID
|
AS_OF = "2026-08-26"
|
PRICE_DATE = "2026-08-25"
|
FINANCE_URL = "https://datacenter-web.eastmoney.com/api/data/v1/get"
|
THS_BASE = "https://basic.10jqka.com.cn"
|
FINANCE_REPORTS = {
|
"main": "RPT_F10_FINANCE_MAINFINADATA",
|
"income": "RPT_DMSK_FN_INCOME",
|
"balance": "RPT_F10_FINANCE_GBALANCE",
|
"cashflow": "RPT_DMSK_FN_CASHFLOW",
|
}
|
THREAD_STATE = threading.local()
|
|
|
def mysql_rows(sql: str) -> list[dict[str, str]]:
|
proc = subprocess.run(
|
[
|
"mysql",
|
"--login-path=ana_semi_admin_preflight",
|
"--default-character-set=utf8mb4",
|
"--batch",
|
"--raw",
|
"-e",
|
sql,
|
],
|
cwd=ROOT,
|
capture_output=True,
|
text=True,
|
encoding="utf-8",
|
errors="strict",
|
check=True,
|
)
|
lines = proc.stdout.splitlines()
|
if not lines:
|
return []
|
reader = csv.DictReader(lines, delimiter="\t")
|
return list(reader)
|
|
|
def universe() -> list[dict[str, Any]]:
|
sql = f"""
|
SELECT s.ticker,s.company,s.market,s.currency,
|
v.valuation_id,v.valuation_date,v.method,v.base_low,v.base_high,
|
v.optimistic_low,v.optimistic_high,v.normalized_profit,
|
v.normalized_pe,v.pb,v.ps,v.consensus_year,v.consensus_profit,
|
v.consensus_count,v.report_path,v.snapshot_path,v.source_hash,
|
p.close,p.trade_date,
|
JSON_UNQUOTE(JSON_EXTRACT(i.raw_json,'$.TotalVolume')) AS static_total_shares
|
FROM stock_valuation.security s
|
JOIN stock_valuation.valuation_version v
|
ON v.ticker=s.ticker AND v.active_to IS NULL
|
LEFT JOIN stock_valuation.daily_price p
|
ON p.ticker=s.ticker AND p.trade_date='{PRICE_DATE}'
|
LEFT JOIN trading_xuntou.cn_stock_instrument_static i
|
ON BINARY i.symbol=BINARY s.ticker
|
WHERE s.active=1
|
ORDER BY s.ticker
|
"""
|
rows = mysql_rows(sql)
|
for row in rows:
|
for key in (
|
"base_low", "base_high", "optimistic_low", "optimistic_high",
|
"normalized_profit", "normalized_pe", "pb", "ps",
|
"consensus_profit", "close", "static_total_shares",
|
):
|
value = row.get(key)
|
row[key] = float(value) if value not in (None, "", "NULL") else None
|
row["consensus_count"] = int(row["consensus_count"]) if row.get("consensus_count") not in (None, "", "NULL") else 0
|
return rows
|
|
|
def session() -> requests.Session:
|
value = getattr(THREAD_STATE, "session", None)
|
if value is None:
|
value = requests.Session()
|
value.headers.update(
|
{
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/140 Safari/537.36",
|
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.6",
|
}
|
)
|
THREAD_STATE.session = value
|
return value
|
|
|
def get(url: str, *, params: dict[str, str] | None = None, referer: str | None = None) -> bytes:
|
headers = {"Referer": referer} if referer else None
|
last: Exception | None = None
|
for attempt in range(3):
|
try:
|
response = session().get(url, params=params, headers=headers, timeout=(8, 25))
|
response.raise_for_status()
|
if not response.content:
|
raise RuntimeError("empty response")
|
return response.content
|
except Exception as exc:
|
last = exc
|
time.sleep(0.5 * (attempt + 1))
|
raise RuntimeError(f"GET failed: {url}: {last}")
|
|
|
def finance_payload(ticker: str, report_name: str) -> tuple[dict[str, Any], str]:
|
params = {
|
"reportName": report_name,
|
"columns": "ALL",
|
"filter": f'(SECUCODE="{ticker}")',
|
"pageNumber": "1",
|
"pageSize": "20",
|
"sortTypes": "-1",
|
"sortColumns": "REPORT_DATE",
|
}
|
body = get(FINANCE_URL, params=params)
|
payload = json.loads(body.decode("utf-8-sig"))
|
return payload, hashlib.sha256(body).hexdigest().upper()
|
|
|
def records(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
return list(((payload.get("result") or {}).get("data") or payload.get("data") or []))
|
|
|
def eligible(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
result = []
|
for row in rows:
|
report_date = str(row.get("REPORT_DATE") or "")[:10]
|
notice_date = str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE") or "")[:10]
|
if report_date and notice_date and report_date <= AS_OF and notice_date <= AS_OF:
|
result.append(row)
|
return result
|
|
|
def select_periods(income_rows: list[dict[str, Any]]) -> tuple[str, str, str]:
|
periods = sorted({str(row["REPORT_DATE"])[:10] for row in eligible(income_rows)})
|
annuals = [period for period in periods if period.endswith("-12-31")]
|
if not annuals:
|
raise RuntimeError("no annual report")
|
annual = annuals[-1]
|
current_candidates = [period for period in periods if period > annual and not period.endswith("-12-31")]
|
if not current_candidates:
|
raise RuntimeError("no current cumulative report")
|
current = current_candidates[-1]
|
prior = f"{int(current[:4]) - 1}{current[4:]}"
|
if prior not in periods:
|
raise RuntimeError(f"no comparative period {prior}")
|
return annual, current, prior
|
|
|
def pick(rows: list[dict[str, Any]], period: str) -> dict[str, Any]:
|
matches = [row for row in eligible(rows) if str(row["REPORT_DATE"])[:10] == period]
|
if not matches:
|
raise RuntimeError(f"missing {period}")
|
return max(matches, key=lambda row: str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE") or ""))
|
|
|
def compact(row: dict[str, Any], keys: tuple[str, ...]) -> dict[str, Any]:
|
return {key: row.get(key) for key in keys if key in row}
|
|
|
INCOME_KEYS = (
|
"REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "SECURITY_CODE", "SECURITY_NAME_ABBR",
|
"TOTAL_OPERATE_INCOME", "TOTALOPERATEREVE", "PARENT_NETPROFIT", "PARENTNETPROFIT",
|
"DEDUCT_PARENT_NETPROFIT", "KCFJCXSYJLR", "BASIC_EPS", "DILUTED_EPS",
|
)
|
CASHFLOW_KEYS = ("REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "NETCASH_OPERATE", "CONSTRUCT_LONG_ASSET")
|
BALANCE_KEYS = (
|
"REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "TOTAL_EQUITY", "TOTAL_EQUITY_PARENT",
|
"MONETARYFUNDS", "MINORITY_EQUITY", "TOTAL_ASSETS", "TOTAL_LIABILITIES",
|
"SHORT_LOAN", "NONCURRENT_LIAB_1YEAR", "LONG_LOAN", "BOND_PAYABLE", "LEASE_LIAB",
|
"SHORT_BOND_PAYABLE", "TRADE_FINASSET_NOTFVTPL", "TRADE_FINASSET", "FVTPL_FINASSET",
|
"APPOINT_FVTPL_FINASSET", "AVAILABLE_SALE_FINASSET", "DERIVE_FINASSET", "BUY_RESALE_FINASSET",
|
)
|
MAIN_KEYS = (
|
"REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "TOTAL_SHARE", "TOTAL_EQUITY_PARENT",
|
"ROEJQ", "XSMLL", "ZCFZL",
|
)
|
|
|
def parse_money(text: str) -> float | None:
|
cleaned = text.replace(",", "").strip()
|
if cleaned in {"", "--", "-"}:
|
return None
|
match = re.search(r"(-?\d+(?:\.\d+)?)\s*([万亿]?)", cleaned)
|
if not match:
|
return None
|
value = float(match.group(1))
|
return value * ({"亿": 1e8, "万": 1e4}.get(match.group(2), 1.0))
|
|
|
def parse_forecast_page(body: bytes) -> tuple[list[dict[str, Any]], str]:
|
text = body.decode("gb18030", errors="replace")
|
soup = BeautifulSoup(text, "lxml")
|
root = soup.find(id="forecastdetail")
|
rows: list[dict[str, Any]] = []
|
if root:
|
table = root.find("table")
|
if table:
|
for tr in table.select("tbody tr"):
|
cells = [cell.get_text(" ", strip=True) for cell in tr.find_all(["th", "td"])]
|
if len(cells) < 9 or not re.fullmatch(r"20\d{2}-\d{2}-\d{2}", cells[-1]):
|
continue
|
rows.append(
|
{
|
"institution": cells[0],
|
"analyst": cells[1],
|
"eps_2026": parse_money(cells[2]),
|
"eps_2027": parse_money(cells[3]),
|
"eps_2028": parse_money(cells[4]),
|
"profit_2026": parse_money(cells[5]),
|
"profit_2027": parse_money(cells[6]),
|
"profit_2028": parse_money(cells[7]),
|
"report_date": cells[8],
|
}
|
)
|
return rows, hashlib.sha256(body).hexdigest().upper()
|
|
|
def extract_after(text: str, label: str, next_labels: tuple[str, ...]) -> str:
|
start = text.find(label)
|
if start < 0:
|
return ""
|
value = text[start + len(label):]
|
stops = [value.find(item) for item in next_labels if value.find(item) >= 0]
|
return value[: min(stops) if stops else len(value)].strip()
|
|
|
def parse_business_page(body: bytes) -> tuple[dict[str, str], str]:
|
text = body.decode("gb18030", errors="replace")
|
soup = BeautifulSoup(text, "lxml")
|
visible = re.sub(r"\s+", " ", soup.get_text(" ", strip=True))
|
main = extract_after(visible, "主营业务：", ("产品类型：", "产品名称：", "经营范围："))
|
product_types = extract_after(visible, "产品类型：", ("产品名称：", "经营范围："))
|
product_names = extract_after(visible, "产品名称：", ("经营范围：", "主营构成分析"))
|
return {
|
"main_business": main,
|
"product_types": product_types,
|
"product_names": product_names,
|
}, hashlib.sha256(body).hexdigest().upper()
|
|
|
def fetch_one(item: dict[str, Any]) -> dict[str, Any]:
|
ticker = item["ticker"]
|
code = ticker.split(".")[0]
|
if ticker.endswith(".HK"):
|
return {"ticker": ticker, "status": "UNSUPPORTED_HK", "errors": ["A股公开源不覆盖港股"]}
|
errors: list[str] = []
|
payloads: dict[str, Any] = {}
|
hashes: dict[str, str] = {}
|
try:
|
for kind, report_name in FINANCE_REPORTS.items():
|
payloads[kind], hashes[f"finance_{kind}"] = finance_payload(ticker, report_name)
|
income_rows = records(payloads["income"])
|
annual, current, prior = select_periods(income_rows)
|
selected = {
|
"annual_income": compact(pick(income_rows, annual), INCOME_KEYS),
|
"current_income": compact(pick(income_rows, current), INCOME_KEYS),
|
"prior_income": compact(pick(income_rows, prior), INCOME_KEYS),
|
"annual_cashflow": compact(pick(records(payloads["cashflow"]), annual), CASHFLOW_KEYS),
|
"current_cashflow": compact(pick(records(payloads["cashflow"]), current), CASHFLOW_KEYS),
|
"prior_cashflow": compact(pick(records(payloads["cashflow"]), prior), CASHFLOW_KEYS),
|
"current_balance": compact(pick(records(payloads["balance"]), current), BALANCE_KEYS),
|
"current_main": compact(pick(records(payloads["main"]), current), MAIN_KEYS),
|
}
|
except Exception as exc:
|
errors.append(f"finance:{type(exc).__name__}:{exc}")
|
selected = {}
|
annual = current = prior = ""
|
# Banks and a few financial issuers can omit the industrial cash-flow
|
# comparative row. Preserve income/balance evidence so their PB model
|
# can still be refreshed without inventing cash flow.
|
try:
|
income_rows = records(payloads.get("income") or {})
|
annual, current, prior = select_periods(income_rows)
|
selected = {
|
"annual_income": compact(pick(income_rows, annual), INCOME_KEYS),
|
"current_income": compact(pick(income_rows, current), INCOME_KEYS),
|
"prior_income": compact(pick(income_rows, prior), INCOME_KEYS),
|
"current_main": compact(pick(records(payloads.get("main") or {}), current), MAIN_KEYS),
|
}
|
balance_rows = records(payloads.get("balance") or {})
|
if balance_rows:
|
selected["current_balance"] = compact(pick(balance_rows, current), BALANCE_KEYS)
|
cash_rows = records(payloads.get("cashflow") or {})
|
if cash_rows:
|
selected.update(
|
{
|
"annual_cashflow": compact(pick(cash_rows, annual), CASHFLOW_KEYS),
|
"current_cashflow": compact(pick(cash_rows, current), CASHFLOW_KEYS),
|
"prior_cashflow": compact(pick(cash_rows, prior), CASHFLOW_KEYS),
|
}
|
)
|
except Exception as partial_exc:
|
errors.append(f"finance_partial:{type(partial_exc).__name__}:{partial_exc}")
|
try:
|
worth = get(f"{THS_BASE}/{code}/worth.html", referer=f"{THS_BASE}/{code}/")
|
forecasts, hashes["institution_detail"] = parse_forecast_page(worth)
|
except Exception as exc:
|
forecasts = []
|
errors.append(f"forecast:{type(exc).__name__}:{exc}")
|
try:
|
operate = get(f"{THS_BASE}/{code}/operate.html", referer=f"{THS_BASE}/{code}/")
|
business, hashes["business_identity"] = parse_business_page(operate)
|
except Exception as exc:
|
business = {"main_business": "", "product_types": "", "product_names": ""}
|
errors.append(f"business:{type(exc).__name__}:{exc}")
|
result = {
|
"ticker": ticker,
|
"company": item["company"],
|
"status": "OK" if not errors else ("PARTIAL" if selected else "FAILED"),
|
"as_of": AS_OF,
|
"periods": {"annual": annual, "current": current, "prior": prior},
|
"selected": selected,
|
"forecasts": [row for row in forecasts if row["report_date"] <= AS_OF],
|
"business": business,
|
"raw_sha256": hashes,
|
"errors": errors,
|
}
|
return result
|
|
|
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 action_inventory() -> None:
|
rows = universe()
|
TMP_ROOT.mkdir(parents=True, exist_ok=True)
|
columns = list(rows[0])
|
with (TMP_ROOT / "universe.csv").open("w", encoding="utf-8", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=columns)
|
writer.writeheader()
|
writer.writerows(rows)
|
write_json(
|
TMP_ROOT / "inventory.json",
|
{
|
"batch_id": BATCH_ID,
|
"as_of": AS_OF,
|
"price_date": PRICE_DATE,
|
"created_at": datetime.now().astimezone().isoformat(),
|
"universe_count": len(rows),
|
"a_share_count": sum(not row["ticker"].endswith(".HK") for row in rows),
|
"price_covered": sum(row["close"] is not None for row in rows),
|
},
|
)
|
print(json.dumps({"universe": len(rows), "price_covered": sum(row["close"] is not None for row in rows)}, ensure_ascii=False))
|
|
|
def action_fetch(workers: int) -> None:
|
rows = universe()
|
RAW_ROOT.mkdir(parents=True, exist_ok=True)
|
completed: list[dict[str, Any]] = []
|
started = time.monotonic()
|
with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="midyear-refresh") as pool:
|
futures = {pool.submit(fetch_one, item): item for item in rows}
|
for index, future in enumerate(as_completed(futures), 1):
|
item = futures[future]
|
try:
|
result = future.result()
|
except Exception as exc:
|
result = {"ticker": item["ticker"], "company": item["company"], "status": "FAILED", "errors": [repr(exc)]}
|
write_json(RAW_ROOT / f"{result['ticker'].replace('.', '_')}.json", result)
|
completed.append(result)
|
if index % 50 == 0 or index == len(rows):
|
print(f"[{index}/{len(rows)}] elapsed={time.monotonic()-started:.1f}s", flush=True)
|
counts: dict[str, int] = {}
|
for row in completed:
|
counts[row["status"]] = counts.get(row["status"], 0) + 1
|
write_json(
|
TMP_ROOT / "fetch_manifest.json",
|
{
|
"batch_id": BATCH_ID,
|
"as_of": AS_OF,
|
"price_date": PRICE_DATE,
|
"finished_at": datetime.now().astimezone().isoformat(),
|
"wall_seconds": round(time.monotonic() - started, 3),
|
"counts": counts,
|
"files": len(completed),
|
},
|
)
|
print(json.dumps(counts, ensure_ascii=False))
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("action", choices=("inventory", "fetch"))
|
parser.add_argument("--workers", type=int, default=16)
|
args = parser.parse_args()
|
if hasattr(sys.stdout, "reconfigure"):
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
if args.action == "inventory":
|
action_inventory()
|
else:
|
action_fetch(args.workers)
|
|
|
if __name__ == "__main__":
|
main()
|
