#!/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()
|
