#!/usr/bin/env python3 """Build a read-only semiconductor MA60 rebound ranking from local MySQL.""" from __future__ import annotations import argparse import csv import hashlib import json import os from collections import defaultdict from datetime import date, datetime from decimal import Decimal from pathlib import Path import pymysql SCOPE_VERSION = "semiconductor-scope-2026-08-07-v1" # 口径：主营或重要业务直接位于芯片设计、制造、封测、EDA，或直接服务晶圆制造/封测的设备材料。 # relation=核心：芯片/晶圆/封测/EDA；上游：半导体专用设备、材料与封装基板；邻接：主题相关但半导体并非主要收入来源。 SCOPE = { "000021.SZ": ("核心（混合业务）", "存储半导体封测"), "001270.SZ": ("核心", "射频芯片"), "002049.SZ": ("核心", "特种集成电路"), "002151.SZ": ("核心（混合业务）", "导航芯片"), "002156.SZ": ("核心", "集成电路封测"), "002185.SZ": ("核心", "集成电路封测"), "002371.SZ": ("上游", "刻蚀/薄膜/清洗设备"), "002402.SZ": ("核心（混合业务）", "相控阵T/R芯片"), "002405.SZ": ("核心（混合业务）", "汽车芯片"), "002409.SZ": ("上游", "半导体材料/电子特气"), "002436.SZ": ("上游", "半导体测试板/封装基板"), "002916.SZ": ("上游", "封装基板"), "300053.SZ": ("核心（混合业务）", "宇航芯片/模块"), "300054.SZ": ("上游", "CMP抛光材料"), "300101.SZ": ("核心（混合业务）", "高端集成电路"), "300139.SZ": ("核心（混合业务）", "集成电路设计"), "300260.SZ": ("上游", "高纯管路/阀门/真空部件"), "300327.SZ": ("核心", "MCU/驱动芯片"), "300346.SZ": ("上游", "前驱体/电子特气/光刻胶"), "300373.SZ": ("核心", "功率半导体"), "300395.SZ": ("上游（混合业务）", "半导体石英材料"), "300456.SZ": ("核心", "MEMS晶圆制造"), "300474.SZ": ("核心（混合业务）", "图形显控芯片"), "300567.SZ": ("上游", "半导体检测设备"), "300576.SZ": ("上游", "光刻胶"), "300604.SZ": ("上游", "半导体测试设备"), "300613.SZ": ("核心", "视频处理芯片"), "300666.SZ": ("上游", "靶材/精密部件"), "300671.SZ": ("核心", "模拟集成电路"), "300672.SZ": ("核心", "多媒体/存储芯片"), "300757.SZ": ("上游（混合业务）", "泛半导体自动化设备"), "301269.SZ": ("核心", "EDA"), "301308.SZ": ("核心", "存储芯片/先进封测"), "301583.SZ": ("上游", "半导体设备精密零部件"), "301611.SZ": ("上游", "半导体陶瓷零部件"), "600206.SH": ("上游（混合业务）", "高纯金属靶材"), "600360.SH": ("核心", "功率半导体IDM"), "600584.SH": ("核心", "集成电路封测"), "600641.SH": ("上游", "离子注入设备"), "600877.SH": ("核心", "特种集成电路"), "603061.SH": ("上游", "测试分选设备"), "603068.SH": ("核心", "无线通信芯片"), "603078.SH": ("上游", "湿电子化学品"), "603650.SH": ("上游", "光刻胶/电子化学品"), "603893.SH": ("核心", "智能应用处理器"), "603936.SH": ("上游", "封装载板"), "605358.SH": ("核心", "硅片/功率器件/射频芯片"), "688002.SH": ("核心", "红外探测器芯片"), "688012.SH": ("上游", "刻蚀/薄膜沉积设备"), "688019.SH": ("上游", "CMP材料"), "688037.SH": ("上游", "涂胶显影/清洗设备"), "688049.SH": ("核心", "智能音频SoC"), "688072.SH": ("上游", "薄膜沉积设备"), "688107.SH": ("核心", "FPGA/EDA"), "688110.SH": ("核心", "存储芯片"), "688120.SH": ("上游", "CMP/减薄设备"), "688126.SH": ("上游", "半导体硅片"), "688135.SH": ("核心", "晶圆/成品测试"), "688153.SH": ("核心", "射频前端芯片"), "688172.SH": ("核心", "半导体制造/IDM"), "688199.SH": ("上游（混合业务）", "半导体化学材料"), "688213.SH": ("核心", "CMOS图像传感器"), "688233.SH": ("上游", "大直径硅材料/硅零部件"), "688262.SH": ("核心", "嵌入式CPU"), "688268.SH": ("上游", "电子特种气体"), "688270.SH": ("核心", "射频芯片/微系统"), "688313.SH": ("核心", "光通信芯片"), "688343.SH": ("核心（混合业务）", "AI推理芯片"), "688347.SH": ("核心", "晶圆代工"), "688352.SH": ("核心", "先进封装与测试"), "688361.SH": ("上游", "量检测设备"), "688362.SH": ("核心", "封装与测试"), "688368.SH": ("核心", "电源管理/控制驱动芯片"), "688372.SH": ("核心", "集成电路测试"), "688375.SH": ("核心", "射频集成电路"), "688380.SH": ("核心", "MCU/控制芯片"), "688385.SH": ("核心", "集成电路设计"), "688401.SH": ("上游", "掩膜版"), "688409.SH": ("上游", "半导体设备精密零部件"), "688416.SH": ("核心", "存储芯片/MCU"), "688419.SH": ("上游", "半导体封装设备"), "688439.SH": ("核心", "高可靠集成电路"), "688458.SH": ("核心", "模拟及数模混合芯片"), "688486.SH": ("核心", "高速混合信号芯片"), "688498.SH": ("核心", "光通信激光器芯片"), "688515.SH": ("核心", "高速有线通信芯片"), "688521.SH": ("核心", "芯片IP/定制"), "688525.SH": ("核心", "存储芯片/先进封测"), "688535.SH": ("上游", "电子封装材料"), "688536.SH": ("核心", "模拟及数模混合芯片"), "688591.SH": ("核心", "低功耗无线物联网芯片"), "688595.SH": ("核心", "ADC/MCU"), "688596.SH": ("上游", "高纯工艺系统/电子气体"), "688627.SH": ("上游", "半导体检测设备"), "688630.SH": ("上游", "直写光刻设备"), "688652.SH": ("上游", "工艺温控/尾气处理设备"), "688702.SH": ("核心", "以太网交换芯片"), "688709.SH": ("核心", "特种集成电路"), "688711.SH": ("核心", "功率半导体"), "688825.SH": ("核心", "DRAM存储芯片制造"), "688981.SH": ("核心", "晶圆代工"), } ADJACENT = { "002008.SZ": ("邻接", "通用激光/部分半导体设备应用"), "002129.SZ": ("邻接", "光伏硅片，非集成电路硅片"), "002222.SZ": ("邻接", "光学晶体/激光器件"), "002338.SZ": ("邻接", "精密光电仪器"), "301021.SZ": ("邻接", "精密激光加工设备"), "301307.SZ": ("邻接", "精密压铸件"), "301392.SZ": ("邻接", "真空镀膜设备"), "301421.SZ": ("邻接", "精密光学"), "688371.SH": ("邻接", "纳米薄膜防护涂层"), "688392.SH": ("邻接", "超声设备/部分半导体应用"), "688502.SH": ("邻接", "高端精密光学"), "688610.SH": ("邻接", "机器视觉核心部件"), } def connect(database: str): required = ["STOCK_MA60_DB_HOST", "STOCK_MA60_DB_PORT", "STOCK_MA60_DB_USER", "STOCK_MA60_DB_PASS"] missing = [name for name in required if not os.getenv(name)] if missing: raise RuntimeError(f"missing database environment variables: {', '.join(missing)}") return pymysql.connect( host=os.environ["STOCK_MA60_DB_HOST"], port=int(os.environ["STOCK_MA60_DB_PORT"]), user=os.environ["STOCK_MA60_DB_USER"], password=os.environ["STOCK_MA60_DB_PASS"], database=database, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=True, ) def decimal_float(value): if value is None: return None if isinstance(value, Decimal): return float(value) return float(value) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest().upper() def fmt_num(value, digits=2): return "—" if value is None else f"{value:.{digits}f}" def fmt_pct(value): return "—" if value is None else f"{value * 100:.2f}%" def load_security_and_valuation(as_of: date): with connect("stock_valuation") as conn, conn.cursor() as cursor: cursor.execute( """ SELECT s.ticker, s.company, v.valuation_id, v.base_low, v.base_high, j.close AS ledger_close, j.label FROM security s JOIN valuation_version v ON v.ticker = s.ticker AND v.active_from <= %s AND (v.active_to IS NULL OR v.active_to >= %s) LEFT JOIN daily_judgement j ON j.ticker = s.ticker AND j.trade_date = %s WHERE s.active = 1 AND s.currency = 'CNY' """, (as_of, as_of, as_of), ) rows = cursor.fetchall() return {row["ticker"]: row for row in rows} def load_calendar_and_bars(as_of: date, tickers: list[str]): with connect("trading_xuntou") as conn, conn.cursor() as cursor: cursor.execute( """ SELECT asset_version, market, calendar_source_kind, calendar_source_version, window_start, window_end, generated_at, status FROM formal_trading_calendar_assets WHERE is_current = 1 AND market = 'SH' ORDER BY generated_at DESC LIMIT 1 """ ) calendar = cursor.fetchone() if not calendar or calendar["status"] != "current" or calendar["window_end"] < as_of: raise RuntimeError("authoritative trading calendar does not cover as-of date") cursor.execute( "SELECT trade_date, is_open FROM formal_trading_calendar_days WHERE market='SH' AND trade_date=%s AND asset_version=%s", (as_of, calendar["asset_version"]), ) day = cursor.fetchone() if not day or int(day["is_open"]) != 1: raise RuntimeError("as-of date is not a proven open trading day") placeholders = ",".join(["%s"] * len(tickers)) query = f""" WITH dedup AS ( SELECT symbol, trade_date, close, amount, source, source_batch_id, updated_at, ROW_NUMBER() OVER (PARTITION BY symbol, trade_date ORDER BY id DESC) AS duplicate_rank FROM cn_stock_kline_1d_front WHERE symbol IN ({placeholders}) AND trade_date <= %s AND close IS NOT NULL AND close > 0 ), ranked AS ( SELECT symbol, trade_date, close, amount, source, source_batch_id, updated_at, ROW_NUMBER() OVER (PARTITION BY symbol ORDER BY trade_date DESC) AS recent_rank FROM dedup WHERE duplicate_rank = 1 ) SELECT symbol, trade_date, close, amount, source, source_batch_id, updated_at FROM ranked WHERE recent_rank <= 60 ORDER BY symbol, trade_date """ cursor.execute(query, (*tickers, as_of)) rows = cursor.fetchall() grouped = defaultdict(list) for row in rows: grouped[row["symbol"]].append(row) return calendar, grouped def calculate(ticker, relation, segment, security, bars, as_of): if not security: return None, {"ticker": ticker, "reason": "不在当前估值台账有效A股范围"} if len(bars) != 60: return None, {"ticker": ticker, "company": security["company"], "reason": f"有效日K不足60条（{len(bars)}条）"} if bars[-1]["trade_date"] != as_of: return None, {"ticker": ticker, "company": security["company"], "reason": f"最新有效日K为{bars[-1]['trade_date']}，不是{as_of}"} closes = [decimal_float(row["close"]) for row in bars] amounts = [decimal_float(row["amount"]) for row in bars] close = closes[-1] ma60 = sum(closes) / 60 below = (ma60 - close) / ma60 rebound = ma60 / close - 1 amount5 = None if any(v is None for v in amounts[-5:]) else sum(amounts[-5:]) / 5 amount20 = None if any(v is None for v in amounts[-20:]) else sum(amounts[-20:]) / 20 amount_ratio = amount5 / amount20 if amount5 is not None and amount20 not in (None, 0) else None if amount_ratio is None: amount_state = "未知" elif amount_ratio >= 1.2: amount_state = "放量" elif amount_ratio <= 0.8: amount_state = "缩量" else: amount_state = "平量" ledger_close = decimal_float(security["ledger_close"]) price_check = ledger_close is None or abs(ledger_close - close) < 0.005 return { "ticker": ticker, "company": security["company"], "relation": relation, "segment": segment, "trade_date": str(as_of), "close": close, "ma60": ma60, "below_ma60_pct": below, "rebound_to_ma60_pct": rebound, "position": "低于MA60" if close < ma60 else "不低于MA60", "amount_5d_20d_ratio": amount_ratio, "amount_state": amount_state, "valuation_label": security["label"] or "未形成当日台账判断", "base_low": decimal_float(security["base_low"]), "base_high": decimal_float(security["base_high"]), "valuation_id": security["valuation_id"], "ledger_price_match": price_check, "kline_source": bars[-1]["source"], "kline_batch": bars[-1]["source_batch_id"], "kline_updated_at": str(bars[-1]["updated_at"]), }, None def write_csv(path: Path, rows: list[dict]): fields = [ "rank", "ticker", "company", "relation", "segment", "trade_date", "close", "ma60", "below_ma60_pct", "rebound_to_ma60_pct", "position", "amount_5d_20d_ratio", "amount_state", "valuation_label", "base_low", "base_high", "valuation_id", "ledger_price_match", "kline_source", "kline_batch", "kline_updated_at", ] with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for index, row in enumerate(rows, 1): item = dict(row) item["rank"] = index if row["position"] == "低于MA60" else "" writer.writerow(item) def table_lines(rows: list[dict], ranked=True): if not rows: return ["无。"] lines = [ "| 排名 | 公司 | 代码 | 产业链位置 | 收盘价 | MA60 | 低于MA60 | 回到MA60空间 | 量能 | 估值判断 | 基准合理区间 |", "|---:|---|---|---|---:|---:|---:|---:|---|---|---:|", ] for index, row in enumerate(rows, 1): rank = str(index) if ranked else "—" lines.append( f"| {rank} | {row['company']} | {row['ticker']} | {row['relation']}·{row['segment']} | " f"{row['close']:.2f} | {row['ma60']:.2f} | {fmt_pct(row['below_ma60_pct'])} | " f"{fmt_pct(row['rebound_to_ma60_pct'])} | {row['amount_state']}({fmt_num(row['amount_5d_20d_ratio'])}×) | " f"{row['valuation_label']} | {row['base_low']:.2f}—{row['base_high']:.2f} |" ) return lines def main(): parser = argparse.ArgumentParser() parser.add_argument("--as-of", default="2026-08-06") parser.add_argument("--output-dir", required=True) args = parser.parse_args() as_of = date.fromisoformat(args.as_of) output_dir = Path(args.output_dir) output_dir.mkdir(parents=True, exist_ok=True) securities = load_security_and_valuation(as_of) requested = sorted(set(SCOPE) | set(ADJACENT)) calendar, grouped = load_calendar_and_bars(as_of, requested) main_rows, adjacent_rows, gaps = [], [], [] for ticker, (relation, segment) in SCOPE.items(): row, gap = calculate(ticker, relation, segment, securities.get(ticker), grouped.get(ticker, []), as_of) if row: main_rows.append(row) else: gaps.append(gap) for ticker, (relation, segment) in ADJACENT.items(): row, gap = calculate(ticker, relation, segment, securities.get(ticker), grouped.get(ticker, []), as_of) if row: adjacent_rows.append(row) else: gaps.append(gap) below = sorted([r for r in main_rows if r["position"] == "低于MA60"], key=lambda r: r["rebound_to_ma60_pct"], reverse=True) above = sorted([r for r in main_rows if r["position"] != "低于MA60"], key=lambda r: r["rebound_to_ma60_pct"], reverse=True) adjacent_below = sorted([r for r in adjacent_rows if r["position"] == "低于MA60"], key=lambda r: r["rebound_to_ma60_pct"], reverse=True) adjacent_above = sorted([r for r in adjacent_rows if r["position"] != "低于MA60"], key=lambda r: r["rebound_to_ma60_pct"], reverse=True) value_intersection = [r for r in below if r["valuation_label"] in {"偏低", "基本合理"}] csv_path = output_dir / f"半导体相关股票低于60日均线排序_{as_of.strftime('%Y%m%d')}.csv" md_path = output_dir / f"半导体相关股票低于60日均线排序_{as_of.strftime('%Y%m%d')}.md" gaps_path = output_dir / f"半导体相关股票MA60计算缺口_{as_of.strftime('%Y%m%d')}.csv" write_csv(csv_path, below + above) with gaps_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=["ticker", "company", "reason"]) writer.writeheader() writer.writerows(gaps) high_space_expensive = [r for r in below if r["valuation_label"] in {"偏贵", "明显偏贵"}] lines = [ f"# 半导体相关股票低于60日均线排序（{as_of}）", "", "> 本表是技术位置筛选，不是目标价或交易指令。MA60只描述过去60个有效交易日的平均成本；价格可能继续下跌，也可能长期不回到MA60。", "", "## 1. 结论摘要", "", f"- 当前估值台账有效A股范围：{len(securities)}只。", f"- 主口径识别半导体核心及直接上游：{len(main_rows)}只；其中低于MA60：{len(below)}只，不低于MA60：{len(above)}只。", f"- 弱相关邻接观察：{len(adjacent_rows)}只；其中低于MA60：{len(adjacent_below)}只。邻接标的不参与主排名。", f"- 计算缺口：{len(gaps)}只。", f"- 主排名中虽然低于MA60、但基本面估值仍为偏贵或明显偏贵：{len(high_space_expensive)}只。", f"- 同时满足低于MA60且估值为偏低/基本合理：{len(value_intersection)}只。", "", "排序按“回到MA60的理论涨幅”从大到小。该指标比“低于MA60的比例”略大，公式分别为：", "", "```text", "低于MA60幅度 = (MA60 - 当前收盘价) / MA60", "回到MA60理论空间 = MA60 / 当前收盘价 - 1", "MA60 = 截至基准日最近60个有效前复权收盘价的算术平均", "```", "", "## 2. 双重筛选：低于MA60且估值不贵", "", "该组同时满足技术位置低于MA60、估值台账判断为偏低或基本合理，仍按回到MA60理论空间排序；它比单看均线距离更有参考价值，但也不代表反弹必然发生。", "", *table_lines(value_intersection), "", "## 3. 主排序：低于MA60的半导体核心及直接上游", "", *table_lines(below), "", "## 4. 不低于MA60的半导体核心及直接上游", "", "这些股票在“反弹到MA60”假设下没有正空间，因此不进入排名。", "", *table_lines(above, ranked=False), "", "## 5. 弱相关邻接观察", "", "该组包括光伏硅片、通用光学、通用激光和真空设备等。它们可能跟随半导体主题波动，但半导体不是主要收入来源或产业归属较弱，故不混入主排名。", "", "### 5.1 低于MA60", "", *table_lines(adjacent_below), "", "### 5.2 不低于MA60", "", *table_lines(adjacent_above, ranked=False), "", "## 6. 使用方法与风险", "", "- “空间最大”只代表距离60日均线最远，不代表反弹概率最高。大幅低于MA60有时恰恰反映盈利下修、行业景气下降、解禁减持、监管或流动性风险。", "- 量能列为最近5日平均成交额/最近20日平均成交额：≥1.20为放量，≤0.80为缩量，其余为平量。缩量下跌后的反弹确认通常弱于放量企稳，但这一指标仍不能单独决策。", "- 估值判断与MA60是两套口径。优先观察“低于MA60且估值为偏低/基本合理”的交集；若估值仍明显偏贵，回到均线只是交易层假设，不代表价值修复。", "- 当前收盘和MA60均来自`trading_xuntou.cn_stock_kline_1d_front`，只读、前复权；基准日由权威交易日历证明为开市日。", "- 行业范围采用固定清单版本，避免每日按关键词漂移。主营变化或新增估值标的时才更新范围。", "", "## 7. 数据与质量信息", "", f"- 行情基准日：{as_of}", f"- 交易日历资产：`{calendar['asset_version']}`，覆盖至{calendar['window_end']}，状态`{calendar['status']}`。", "- 行情表：`trading_xuntou.cn_stock_kline_1d_front`；价格口径：前复权日K收盘价。", f"- 行业范围版本：`{SCOPE_VERSION}`。", f"- 价格与当日估值台账不一致：{sum(not r['ledger_price_match'] for r in main_rows + adjacent_rows)}只。", f"- 缺口明细：`{gaps_path.name}`。", ] md_path.write_text("\n".join(lines) + "\n", encoding="utf-8") manifest = { "generated_at": datetime.now().astimezone().isoformat(), "as_of": str(as_of), "scope_version": SCOPE_VERSION, "universe_a_share_count": len(securities), "main_scope_requested": len(SCOPE), "main_scope_calculated": len(main_rows), "main_below_ma60": len(below), "main_at_or_above_ma60": len(above), "adjacent_scope_requested": len(ADJACENT), "adjacent_scope_calculated": len(adjacent_rows), "adjacent_below_ma60": len(adjacent_below), "gaps": gaps, "calendar": {key: str(value) for key, value in calendar.items()}, "sources": { "market": "trading_xuntou.cn_stock_kline_1d_front", "valuation": "stock_valuation.valuation_version + daily_judgement", }, "outputs": { md_path.name: {"bytes": md_path.stat().st_size, "sha256": sha256(md_path)}, csv_path.name: {"bytes": csv_path.stat().st_size, "sha256": sha256(csv_path)}, gaps_path.name: {"bytes": gaps_path.stat().st_size, "sha256": sha256(gaps_path)}, }, } manifest_path = output_dir / "manifest.json" manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps(manifest, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()