#!/usr/bin/env python3 """Stream a full A-share daily archive into a leakage-aware portfolio study. The main TuShare tarball is scanned once. Parquet members are read from memory and never extracted as a full dataset. The optimizer uses close-of-day information to form a pending target that can only execute at the next session open. """ from __future__ import annotations import argparse import collections import datetime as dt import json import math import re import statistics import subprocess import sys import tarfile import tempfile import zipfile from array import array from dataclasses import dataclass, field from pathlib import Path PORTAL_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PORTAL_ROOT / ".python-deps")) import pyarrow.parquet as pq # noqa: E402 DEFAULT_QUANT_ROOT = Path("/Users/zhuboyu/Desktop/龙门计划/投资/Quant") DEFAULT_TUSHARE = DEFAULT_QUANT_ROOT / "tushare_15000_history_by_api_packages_20260627" DEFAULT_MINUTE = DEFAULT_QUANT_ROOT / "【53】2026A股分钟日频-持续更新到年底" DEFAULT_DAILY = DEFAULT_QUANT_ROOT / "【49】A股5800股票日线持续更新" def is_a_share(code: str) -> bool: return bool( re.match(r"^(?:00[0123]|30[01])\d{3}\.SZ$", code) or re.match(r"^(?:60[0135]|68[89])\d{3}\.SH$", code) or re.match(r"^(?:[48]\d{5}|92\d{4})\.BJ$", code) ) def date_text(raw: str) -> str: return f"{raw[:4]}-{raw[4:6]}-{raw[6:8]}" def safe_float(value, default=0.0) -> float: try: number = float(value) return number if math.isfinite(number) else default except (TypeError, ValueError): return default def annualized_volatility(returns) -> float: if len(returns) < 20: return 0.0 return statistics.stdev(returns) * math.sqrt(252) def limit_fraction(code: str, date: str) -> float: if code.endswith(".BJ"): return 0.30 prefix = code[:3] if prefix in {"688", "689"}: return 0.20 if prefix in {"300", "301"} and date >= "2020-08-24": return 0.20 return 0.10 @dataclass class SymbolState: closes: collections.deque = field(default_factory=lambda: collections.deque(maxlen=201)) returns: collections.deque = field(default_factory=lambda: collections.deque(maxlen=60)) amounts: collections.deque = field(default_factory=lambda: collections.deque(maxlen=20)) normalized_close: float = 0.0 raw_close: float = 0.0 raw_date: str = "" days: int = 0 pe_ttm: float = 0.0 pb: float = 0.0 dividend_yield: float = 0.0 total_market_value: float = 0.0 turnover_rate: float = 0.0 @dataclass(frozen=True) class Config: family: str lookback: int volatility_penalty: float rebalance_days: int top_k: int exposure: float minimum_breadth: float @property def identifier(self): return ( f"LOCAL-{self.family}-L{self.lookback}-VP{self.volatility_penalty:g}" f"-R{self.rebalance_days}-K{self.top_k}-E{self.exposure:g}-B{self.minimum_breadth:g}" ) @dataclass class Simulation: config: Config active: dict[str, float] = field(default_factory=dict) pending: dict[str, float] | None = None returns: array = field(default_factory=lambda: array("f")) daily_turnover: array = field(default_factory=lambda: array("f")) equity: float = 1.0 peak: float = 1.0 cooldown_until: int = -1 cooldown_armed: bool = True turnover: float = 0.0 blocked_orders: int = 0 def config_grid(): values = [] for family in ("dual", "low-vol-trend", "multi-factor", "value-low-vol"): for lookback in (20, 60, 120, 200): for penalty in (0.5, 1.0): for rebalance in (5, 10): for top_k in (20, 40): for exposure in (0.30, 0.50): for breadth in (0.35, 0.50): values.append(Config(family, lookback, penalty, rebalance, top_k, exposure, breadth)) return values def metric(returns, dates, start: str, end: str): values = [float(value) for value, date in zip(returns, dates) if start <= date <= end] if not values: return {"observations": 0, "annualizedReturn": 0, "sharpe": 0, "maxDrawdown": 0, "dailyWinRate": 0, "positiveYearRate": 0} equity = peak = 1.0 max_drawdown = 0.0 years = collections.defaultdict(lambda: 1.0) for value, date in zip((float(value) for value, date in zip(returns, dates) if start <= date <= end), (date for date in dates if start <= date <= end)): equity *= max(0.01, 1 + value) peak = max(peak, equity) max_drawdown = min(max_drawdown, equity / peak - 1) years[date[:4]] *= 1 + value annualized_vol = statistics.stdev(values) * math.sqrt(252) if len(values) > 1 else 0 annualized_return = equity ** (252 / max(1, len(values))) - 1 if equity > 0 else -1 return { "observations": len(values), "totalReturn": round(equity - 1, 6), "annualizedReturn": round(annualized_return, 6), "annualizedVolatility": round(annualized_vol, 6), "sharpe": round(statistics.mean(values) * 252 / annualized_vol, 6) if annualized_vol else 0, "maxDrawdown": round(max_drawdown, 6), "dailyWinRate": round(sum(value > 0 for value in values) / len(values), 6), "positiveYearRate": round(sum(value > 1 for value in years.values()) / max(1, len(years)), 6), "yearly": {year: round(value - 1, 6) for year, value in sorted(years.items())}, } def score_metric(value): return value["annualizedReturn"] + 0.12 * value["sharpe"] - 0.80 * abs(value["maxDrawdown"]) def update_portfolio(sim: Simulation, rows: dict[str, dict], date: str, date_index: int, cost_bps: float): gross_return = 0.0 factors = {} for code, weight in sim.active.items(): row = rows.get(code) factor = row["openFactor"] if row else 1.0 factors[code] = factor gross_return += weight * (factor - 1) if sim.active and 1 + gross_return > 0: sim.active = {code: weight * factors.get(code, 1.0) / (1 + gross_return) for code, weight in sim.active.items()} turnover = 0.0 if sim.pending is not None: executed = dict(sim.active) for code in set(sim.active) | set(sim.pending): old = sim.active.get(code, 0.0) target = sim.pending.get(code, 0.0) row = rows.get(code) blocked = row is None if row is not None: threshold = limit_fraction(code, date) * 0.985 open_gap = row["rawOpen"] / row["preClose"] - 1 if row["preClose"] > 0 else 0 blocked = (target > old and open_gap >= threshold) or (target < old and open_gap <= -threshold) if blocked: sim.blocked_orders += 1 continue if target > 1e-9: executed[code] = target else: executed.pop(code, None) for code in set(sim.active) | set(executed): turnover += abs(sim.active.get(code, 0.0) - executed.get(code, 0.0)) sim.active = executed sim.pending = None net_return = max(-0.25, gross_return - turnover * cost_bps / 10_000) sim.turnover += turnover sim.returns.append(net_return) sim.daily_turnover.append(turnover) sim.equity *= 1 + net_return sim.peak = max(sim.peak, sim.equity) drawdown = sim.equity / sim.peak - 1 if not sim.cooldown_armed and date_index > sim.cooldown_until: sim.peak = sim.equity sim.cooldown_armed = True drawdown = 0.0 if sim.cooldown_armed and drawdown <= -0.08: sim.cooldown_until = date_index + 20 sim.cooldown_armed = False def build_all_rankings(states: dict[str, SymbolState]): families = ("dual", "low-vol-trend", "multi-factor", "value-low-vol") keys = [(lookback, penalty, family) for lookback in (20, 60, 120, 200) for penalty in (0.5, 1.0) for family in families] candidates = {key: [] for key in keys} breadth = {lookback: [0, 0] for lookback in (20, 60, 120, 200)} base_by_lookback = {lookback: [] for lookback in (20, 60, 120, 200)} for code, state in states.items(): if state.days < 120 or len(state.closes) <= 60: continue current = state.closes[-1] long_past = state.closes[0] if len(state.closes) < 201 else state.closes[-201] short_past = state.closes[-21] if min(current, long_past, short_past) <= 0: continue average_amount = statistics.mean(state.amounts) if state.amounts else 0 if average_amount < 50_000: continue vol = annualized_volatility(state.returns) if not 0.02 <= vol <= 1.20: continue if state.total_market_value and state.total_market_value < 200_000: continue if state.turnover_rate > 30: continue long_momentum = current / long_past - 1 short_momentum = current / short_past - 1 for lookback in (20, 60, 120, 200): if len(state.closes) <= lookback: continue past = state.closes[-lookback - 1] if past <= 0: continue momentum = current / past - 1 breadth[lookback][1] += 1 breadth[lookback][0] += long_momentum > 0 if long_momentum <= 0 or momentum <= -0.12: continue value = -math.log(max(0.2, min(20, state.pb))) if state.pb > 0 else 0.0 earnings_yield = 1 / state.pe_ttm if 0 < state.pe_ttm < 200 else 0.0 dividend = max(0.0, min(15.0, state.dividend_yield)) / 100 base_by_lookback[lookback].append({ "code": code, "vol": vol, "momentum": momentum, "long": long_momentum, "short": short_momentum, "value": value + 3 * earnings_yield, "dividend": dividend, "lowvol": -vol, }) def zscores(items, name): values = [item[name] for item in items] average = statistics.mean(values) if values else 0 deviation = statistics.pstdev(values) if len(values) > 1 else 1 return [(value - average) / max(1e-9, deviation) for value in values] for lookback, items in base_by_lookback.items(): factor_z = {name: zscores(items, name) for name in ("momentum", "long", "value", "dividend", "lowvol")} for index, item in enumerate(items): for penalty in (0.5, 1.0): risk = max(0.08, item["vol"]) ** penalty candidates[(lookback, penalty, "dual")].append(((0.55 * item["momentum"] + 0.45 * item["long"]) / risk, item["code"], item["vol"])) candidates[(lookback, penalty, "low-vol-trend")].append(((0.35 * item["momentum"] + 0.55 * item["long"] - 0.10 * max(0, item["short"])) / risk - 0.25 * item["vol"], item["code"], item["vol"])) multi = 0.30 * factor_z["momentum"][index] + 0.15 * factor_z["long"][index] + 0.25 * factor_z["value"][index] + 0.10 * factor_z["dividend"][index] + 0.20 * factor_z["lowvol"][index] value_lowvol = 0.35 * factor_z["value"][index] + 0.20 * factor_z["dividend"][index] + 0.30 * factor_z["lowvol"][index] + 0.15 * factor_z["long"][index] candidates[(lookback, penalty, "multi-factor")].append((multi / risk, item["code"], item["vol"])) candidates[(lookback, penalty, "value-low-vol")].append((value_lowvol / risk, item["code"], item["vol"])) return { key: (breadth[key[0]][0] / max(1, breadth[key[0]][1]), sorted(values, reverse=True)[:200]) for key, values in candidates.items() } def extract_parquet_members(archive: Path, root: Path): with tarfile.open(archive, "r:gz") as tar: members = [member for member in tar.getmembers() if member.isfile() and member.name.endswith(".parquet")] tar.extractall(root, members=members, filter="data") def tar_parquet_rows(archive: Path, daily_basic_archive: Path): # The vendor tar member order is not chronological. Temporarily stage only # this 0.9 GB daily sub-archive, sort the date-partition filenames, and let # TemporaryDirectory remove the staging files after the run. with tempfile.TemporaryDirectory(prefix="quant-atlas-daily-") as temporary: root = Path(temporary) extract_parquet_members(archive, root) extract_parquet_members(daily_basic_archive, root) paths = sorted(root.glob("**/daily__trade_date=*.parquet"), key=lambda path: re.search(r"trade_date=(\d{8})", path.name).group(1)) basic_paths = {re.search(r"trade_date=(\d{8})", path.name).group(1): path for path in root.glob("**/daily_basic__trade_date=*.parquet")} for path in paths: raw_date = re.search(r"trade_date=(\d{8})", path.name).group(1) table = pq.read_table(path, columns=["ts_code", "trade_date", "open", "close", "pre_close", "pct_chg", "amount"]) basic_path = basic_paths.get(raw_date) basic = pq.read_table(basic_path, columns=["ts_code", "pe_ttm", "pb", "dv_ttm", "total_mv", "turnover_rate"]).to_pydict() if basic_path else {} yield raw_date, table.to_pydict(), basic def crosscheck_daily_zip(archive: Path, latest_raw: dict[str, tuple[str, float]], limit=200): comparisons = [] negative_adjusted_prices = 0 with zipfile.ZipFile(archive) as zf: entries = sorted(name for name in zf.namelist() if re.match(r"\d{6}_daily\.csv$", Path(name).name))[:limit] for name in entries: code = Path(name).name[:6] suffix = ".SH" if code.startswith(("6", "9")) else ".BJ" if code.startswith(("4", "8")) else ".SZ" lines = zf.read(name).decode("utf-8-sig", errors="replace").splitlines() if len(lines) < 2: continue header = lines[0].split(",") values = lines[-1].split(",") record = dict(zip(header, values)) date = record.get("日期", "") close = safe_float(record.get("收盘")) reference = latest_raw.get(code + suffix) if reference and reference[0] == date and reference[1] > 0: comparisons.append(abs(close / reference[1] - 1)) qfq = archive.with_name("daily_qfq.zip") if qfq.exists(): with zipfile.ZipFile(qfq) as zf: for name in sorted(zf.namelist())[:50]: lines = zf.read(name).decode("utf-8-sig", errors="replace").splitlines()[1:250] for line in lines: parts = line.split(",") if len(parts) > 3 and safe_float(parts[3]) <= 0: negative_adjusted_prices += 1 return { "filesSampled": limit, "sameDateComparisons": len(comparisons), "medianAbsoluteCloseDifference": round(statistics.median(comparisons), 8) if comparisons else None, "negativeAdjustedPriceRowsFoundInSample": negative_adjusted_prices, "role": "CROSSCHECK_ONLY_NOT_PRIMARY_RETURN_CHAIN", } def main(): parser = argparse.ArgumentParser() parser.add_argument("--tushare-root", type=Path, default=DEFAULT_TUSHARE) parser.add_argument("--minute-root", type=Path, default=DEFAULT_MINUTE) parser.add_argument("--daily-root", type=Path, default=DEFAULT_DAILY) parser.add_argument("--result", type=Path, required=True) parser.add_argument("--report", type=Path, required=True) parser.add_argument("--cost-bps", type=float, default=15.0) args = parser.parse_args() daily_tar = args.tushare_root / "daily" / "daily.tar.gz" daily_basic_tar = args.tushare_root / "daily_basic" / "daily_basic.tar.gz" third_daily = args.daily_root / "8.3日更新" / "A股-日线" / "A股数据_zip" / "daily.zip" minute_7z = args.minute_root / "2026" / "1min.7z" configs = config_grid() simulations = [Simulation(config) for config in configs] simulation_by_config = {sim.config: sim for sim in simulations} states: dict[str, SymbolState] = {} dates = [] total_rows = 0 total_basic_rows = 0 latest_raw: dict[str, tuple[str, float]] = {} for date_index, (raw_date, columns, basic_columns) in enumerate(tar_parquet_rows(daily_tar, daily_basic_tar)): date = date_text(raw_date) dates.append(date) rows = {} basic_by_code = {} if basic_columns: for basic_index, basic_code in enumerate(basic_columns["ts_code"]): basic_by_code[basic_code] = { "pe_ttm": safe_float(basic_columns["pe_ttm"][basic_index]), "pb": safe_float(basic_columns["pb"][basic_index]), "dv_ttm": safe_float(basic_columns["dv_ttm"][basic_index]), "total_mv": safe_float(basic_columns["total_mv"][basic_index]), "turnover_rate": safe_float(basic_columns["turnover_rate"][basic_index]), } total_basic_rows += 1 for index, code in enumerate(columns["ts_code"]): if not is_a_share(code): continue raw_open = safe_float(columns["open"][index]) raw_close = safe_float(columns["close"][index]) pre_close = safe_float(columns["pre_close"][index]) pct_change = safe_float(columns["pct_chg"][index]) / 100 amount = safe_float(columns["amount"][index]) state = states.setdefault(code, SymbolState()) previous_normalized = state.normalized_close normalized_open = previous_normalized * raw_open / pre_close if previous_normalized > 0 and pre_close > 0 else raw_open normalized_close = previous_normalized * (1 + pct_change) if previous_normalized > 0 else raw_close prior_open = getattr(state, "last_open", 0.0) rows[code] = { "rawOpen": raw_open, "preClose": pre_close, "openFactor": normalized_open / prior_open if prior_open > 0 and normalized_open > 0 else 1.0, } state.last_open = normalized_open state.normalized_close = normalized_close state.raw_close = raw_close state.raw_date = date state.days += 1 state.closes.append(normalized_close) state.returns.append(pct_change) state.amounts.append(amount) basic = basic_by_code.get(code) if basic: state.pe_ttm = basic["pe_ttm"] state.pb = basic["pb"] state.dividend_yield = basic["dv_ttm"] state.total_market_value = basic["total_mv"] state.turnover_rate = basic["turnover_rate"] latest_raw[code] = (date, raw_close) total_rows += 1 for sim in simulations: update_portfolio(sim, rows, date, date_index, args.cost_bps) if date_index >= 201 and date_index % 5 == 0: ranking_cache = build_all_rankings(states) for config in configs: if date_index % config.rebalance_days: continue key = (config.lookback, config.volatility_penalty, config.family) breadth, ranking = ranking_cache[key] sim = simulation_by_config[config] if date_index <= sim.cooldown_until or breadth < config.minimum_breadth: sim.pending = {} continue selected = ranking[: config.top_k] exposure = config.exposure * (0.5 if sim.equity / sim.peak - 1 <= -0.04 else 1.0) if not selected: sim.pending = {} continue inverse_vol = [1 / max(0.08, item[2]) for item in selected] divisor = sum(inverse_vol) sim.pending = {item[1]: min(0.05, exposure * inverse_vol[index] / divisor) for index, item in enumerate(selected)} if (date_index + 1) % 500 == 0: print(f"streamed {date_index + 1} sessions, {total_rows:,} rows, {len(states):,} symbols", flush=True) ranges = { "calibration": {"start": "2000-01-01", "end": "2013-12-31"}, "validation": {"start": "2014-01-01", "end": "2021-12-31"}, "finalHoldout": {"start": "2022-01-01", "end": dates[-1]}, "fullResearch": {"start": dates[0], "end": dates[-1]}, } evaluated = [] for sim in simulations: item = {"config": sim.config.__dict__, "id": sim.config.identifier, "turnover": round(sim.turnover, 4), "blockedOrders": sim.blocked_orders} for name, span in ranges.items(): item[name] = metric(sim.returns, dates, span["start"], span["end"]) evaluated.append(item) calibration_top = sorted(evaluated, key=lambda item: score_metric(item["calibration"]), reverse=True)[:40] winner = max(calibration_top, key=lambda item: score_metric(item["validation"])) validation_passers = [item for item in calibration_top if item["validation"]["annualizedReturn"] > 0.04 and item["validation"]["sharpe"] > 0.45 and item["validation"]["maxDrawdown"] > -0.20] validation_top = sorted(calibration_top, key=lambda item: score_metric(item["validation"]), reverse=True)[:5] ensemble_sims = [simulation_by_config[next(config for config in configs if config.identifier == item["id"])] for item in validation_top] ensemble_returns = array("f", ( statistics.mean(float(sim.returns[index]) for sim in ensemble_sims) for index in range(len(dates)) )) ensemble_turnover = array("f", ( statistics.mean(float(sim.daily_turnover[index]) for sim in ensemble_sims) for index in range(len(dates)) )) validation_ensemble = { "id": "LOCAL-VALIDATION-TOP5-EQUAL-ENSEMBLE", "memberIds": [item["id"] for item in validation_top], "selection": "仅使用校准入围与验证期排名；最终留出不参与成员选择", "turnover": round(sum(ensemble_turnover), 4), } for name, span in ranges.items(): validation_ensemble[name] = metric(ensemble_returns, dates, span["start"], span["end"]) winner_sim = simulation_by_config[next(config for config in configs if config.identifier == winner["id"])] double_cost_returns = array("f", ( float(value) - float(turnover) * args.cost_bps / 10_000 for value, turnover in zip(winner_sim.returns, winner_sim.daily_turnover) )) winner_double_cost = { name: metric(double_cost_returns, dates, span["start"], span["end"]) for name, span in ranges.items() } recommended = validation_ensemble if score_metric(validation_ensemble["validation"]) > score_metric(winner["validation"]) else winner crosscheck = crosscheck_daily_zip(third_daily, latest_raw) listing = subprocess.run( ["tar", "-tf", str(minute_7z)], check=True, capture_output=True, text=True, ) names = listing.stdout.splitlines() minute_meta = { "archive": str(minute_7z), "entries": len(names), "aShareMinuteFiles": sum(name.endswith(".csv") for name in names), "containsTargetEtfs": {code: any(code in name for name in names) for code in ("510900", "511010", "518880")}, "coverage": "2026_ONLY", "role": "A_SHARE_EXECUTION_CALIBRATION_ONLY", } generated_at = dt.datetime.now(dt.timezone.utc).isoformat() result = { "version": "local-archive-streaming-optimizer-v1", "generatedAt": generated_at, "mode": "RESEARCH_ONLY_REAL_ORDERS_LOCKED", "dataset": { "primary": str(daily_tar), "pointInTimeDailyValuation": str(daily_basic_tar), "sessions": len(dates), "start": dates[0], "end": dates[-1], "rows": total_rows, "dailyValuationRows": total_basic_rows, "securitiesObserved": len(states), "fullArchiveExtracted": False, "streamingRead": True, "temporaryDailySubarchiveStaging": True, "temporaryStagingDeletedAfterRun": True, }, "method": { "selection": "先按校准期排序保留40组，只使用验证期选出冠军；2022年后最终留出集不参与选参", "execution": "T日收盘后形成目标，T+1开盘执行；普通A股不做日内回转", "costBpsPerTurnover": args.cost_bps, "survivorshipBias": "MITIGATED_BY_DATE_PARTITIONED_TUSHARE_DAILY", "guarantee": False, }, "ranges": ranges, "configsEvaluated": len(evaluated), "validationPassersAmongCalibrationTop40": len(validation_passers), "winner": winner, "validationSelectedEnsemble": validation_ensemble, "winnerDoubleCostStress": winner_double_cost, "recommendedChallengerId": recommended["id"], "topValidationCandidates": sorted(calibration_top, key=lambda item: score_metric(item["validation"]), reverse=True)[:10], "secondaryDailyCrosscheck": crosscheck, "minuteArchive": minute_meta, "gates": { "allowedForRealCapital": False, "blockers": [ "本地分钟包只含2026年A股个股，未发现目标T+0 ETF分钟文件", "没有券商Level-2队列、IOPV与真实成交回报", "第三方前复权样本出现非正价格，只能作为交叉核验", "至少需要新的前向影子交易期和人工批准，禁止历史结果自动晋级实盘", ], }, } args.result.parent.mkdir(parents=True, exist_ok=True) args.result.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") pct = lambda value: f"{value * 100:.2f}%" report = [ "# 本地三数据目录流式研究与策略优化 v1", "", f"- 主样本：{len(dates):,}个交易日、{total_rows:,}条A股日线、{len(states):,}只历史出现证券，{dates[0]} 至 {dates[-1]}。", "- 读取方式：只把日线与每日估值两个必要子压缩包临时展开并按日期排序；运行后自动清理，没有解压或复制整套本地资料。", f"- 参数：共{len(evaluated)}组；校准期2000–2013，验证期2014–2021，最终留出期2022–{dates[-1][:4]}。", "- 选参纪律：最终留出期不参与选参；结果不是收益承诺。", "", "## 冻结冠军", "", f"- 配置：`{winner['id']}`。", f"- 验证期：年化{pct(winner['validation']['annualizedReturn'])}，Sharpe {winner['validation']['sharpe']:.2f}，最大回撤{pct(winner['validation']['maxDrawdown'])}。", f"- 最终留出：年化{pct(winner['finalHoldout']['annualizedReturn'])}，Sharpe {winner['finalHoldout']['sharpe']:.2f}，最大回撤{pct(winner['finalHoldout']['maxDrawdown'])}。", f"- 留出期日胜率：{pct(winner['finalHoldout']['dailyWinRate'])}；正收益年份占比{pct(winner['finalHoldout']['positiveYearRate'])}。", f"- 2倍成本压力下留出：年化{pct(winner_double_cost['finalHoldout']['annualizedReturn'])}，最大回撤{pct(winner_double_cost['finalHoldout']['maxDrawdown'])}。", "", "## 仅用验证期选择的前五等权集成", "", f"- 推荐挑战者：`{recommended['id']}`；最终留出不参与单模型或集成成员选择。", f"- 集成验证期：年化{pct(validation_ensemble['validation']['annualizedReturn'])}，Sharpe {validation_ensemble['validation']['sharpe']:.2f}，最大回撤{pct(validation_ensemble['validation']['maxDrawdown'])}。", f"- 集成最终留出：年化{pct(validation_ensemble['finalHoldout']['annualizedReturn'])}，Sharpe {validation_ensemble['finalHoldout']['sharpe']:.2f}，最大回撤{pct(validation_ensemble['finalHoldout']['maxDrawdown'])}。", "", "## 三个目录的用途", "", "- TuShare 15000积分包：主研究源，按交易日保留历史出现股票，显著降低幸存者偏差。", f"- 2026分钟包：{minute_meta['aShareMinuteFiles']:,}个A股个股分钟文件；目标ETF存在性：{minute_meta['containsTargetEtfs']}。仅用于执行摩擦校准，不能替代ETF回测。", f"- 5800股票日线包：抽样发现{crosscheck['negativeAdjustedPriceRowsFoundInSample']}条非正前复权价格，降级为交叉验证源。", "", "## 真实资金边界", "", "策略继续保持研究/镜像状态。没有券商Level-2、IOPV、真实队列与前向验证时，不因历史收益较高而自动启用真实资金。", "", ] args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text("\n".join(report), encoding="utf-8") print(json.dumps({"dataset": result["dataset"], "configsEvaluated": len(evaluated), "winner": winner, "minuteArchive": minute_meta, "crosscheck": crosscheck}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()