from __future__ import annotations import argparse import csv import hashlib from collections import defaultdict from decimal import Decimal, InvalidOperation, getcontext from pathlib import Path getcontext().prec = 38 CASE_ID = "ANA-ROBOT-INDUSTRY-001" ACTION_ID = "NEXT-ROBOT-036" RAW_START = "2026-03-20" ANALYSIS_START = "2026-04-24" ANALYSIS_END = "2026-07-24" REQUIRED_COLUMNS = [ "company_market_id", "symbol", "company_id", "canonical_name", "universe_layer", "priority_bucket", "trade_date", "open", "high", "low", "close", "pre_close", "volume", "amount", "turnover", "source_batch_id", "source_latest_trade_date", "raw_window_start", "analysis_window_start", "analysis_window_end", "analysis_window_flag", "project_conclusion_strength", "formal_pool_effect", ] OUTPUT_COLUMNS = [ "case_id", "action_id", "run_id", "company_market_id", "symbol", "company_id", "canonical_name", "universe_layer", "priority_bucket", "price_rows", "close_computable_days", "daily_return_computable_days", "amount_ratio_computable_days", "first_trade_date", "last_trade_date", "first_close", "last_close", "period_return_pct", "strong_up_days", "max_amount_ratio_20", "metric_completeness_status", "manifestation_type", "gap_reason", "source_latest_trade_date", "currentity_status", "raw_snapshot_sha256", "raw_snapshot_bytes", "raw_snapshot_rows", "summary_derivation_version", "conclusion_strength", "review_status", "formal_pool_effect", ] def decimal_or_none(value: str | None) -> Decimal | None: if value is None or value.strip() == "": return None try: return Decimal(value) except InvalidOperation as exc: raise ValueError(f"invalid decimal value: {value!r}") from exc def decimal_text(value: Decimal | None) -> str: if value is None: return "" rendered = format(value.quantize(Decimal("0.000001")), "f") return rendered.rstrip("0").rstrip(".") if "." in rendered else rendered def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--raw", required=True) parser.add_argument("--output", required=True) parser.add_argument("--run-id", required=True) args = parser.parse_args() raw_path = Path(args.raw) output_path = Path(args.output) raw_bytes = raw_path.read_bytes() raw_sha256 = sha256_bytes(raw_bytes) with raw_path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) if reader.fieldnames != REQUIRED_COLUMNS: raise ValueError(f"raw schema mismatch: {reader.fieldnames!r}") raw_rows = list(reader) if not raw_rows: raise ValueError("raw snapshot is empty") if len(raw_rows) != len({(row["symbol"], row["trade_date"]) for row in raw_rows}): raise ValueError("duplicate symbol/trade_date in raw snapshot") if {row["raw_window_start"] for row in raw_rows} != {RAW_START}: raise ValueError("raw_window_start mismatch") if {row["analysis_window_start"] for row in raw_rows} != {ANALYSIS_START}: raise ValueError("analysis_window_start mismatch") if {row["analysis_window_end"] for row in raw_rows} != {ANALYSIS_END}: raise ValueError("analysis_window_end mismatch") source_latest_values = {row["source_latest_trade_date"] for row in raw_rows} if len(source_latest_values) != 1: raise ValueError("source_latest_trade_date is not constant") source_latest = next(iter(source_latest_values)) if source_latest < ANALYSIS_END: raise ValueError("source currentity gate not met") grouped: dict[str, list[dict[str, str]]] = defaultdict(list) for row in raw_rows: grouped[row["symbol"]].append(row) if len(grouped) != 100: raise ValueError(f"expected 100 symbols, found {len(grouped)}") outputs: list[dict[str, str | int]] = [] for symbol in sorted(grouped): rows = sorted(grouped[symbol], key=lambda row: row["trade_date"]) if rows[-1]["trade_date"] < ANALYSIS_END: raise ValueError(f"symbol currentity gate not met: {symbol}") enriched: list[dict[str, object]] = [] prior_amounts: list[Decimal] = [] prior_close: Decimal | None = None for row in rows: close = decimal_or_none(row["close"]) pre_close = decimal_or_none(row["pre_close"]) amount = decimal_or_none(row["amount"]) base_close = pre_close if pre_close not in (None, Decimal(0)) else prior_close daily_return = None if close is not None and base_close not in (None, Decimal(0)): daily_return = (close / base_close - Decimal(1)) * Decimal(100) amount_ratio = None if amount is not None and len(prior_amounts) >= 20: baseline = sum(prior_amounts[-20:]) / Decimal(20) if baseline != 0: amount_ratio = amount / baseline enriched.append( { "row": row, "close": close, "daily_return": daily_return, "amount_ratio": amount_ratio, } ) if amount is not None: prior_amounts.append(amount) if close is not None: prior_close = close analysis = [ item for item in enriched if ANALYSIS_START <= item["row"]["trade_date"] <= ANALYSIS_END ] first = analysis[0] if analysis else None last = analysis[-1] if analysis else None price_rows = len(analysis) close_days = sum(item["close"] is not None for item in analysis) return_days = sum(item["daily_return"] is not None for item in analysis) amount_days = sum(item["amount_ratio"] is not None for item in analysis) complete = ( price_rows > 0 and close_days == price_rows and return_days == price_rows and amount_days == price_rows ) first_close = first["close"] if first else None last_close = last["close"] if last else None period_return = None if first_close not in (None, Decimal(0)) and last_close is not None: period_return = (last_close / first_close - Decimal(1)) * Decimal(100) strong_up_days = sum( item["daily_return"] is not None and item["daily_return"] >= Decimal("9.5") for item in analysis ) amount_ratios = [ item["amount_ratio"] for item in analysis if item["amount_ratio"] is not None ] max_amount_ratio = max(amount_ratios) if amount_ratios else None if not complete: manifestation = "INSUFFICIENT_DATA" gap_reason = "INCOMPLETE_CLOSE_RETURN_OR_PRIOR20_AMOUNT_BASELINE" elif ( period_return is not None and period_return >= Decimal(20) or strong_up_days >= 1 or max_amount_ratio is not None and max_amount_ratio >= Decimal(2) ): manifestation = "STRONG_MANIFESTATION" gap_reason = "NONE" else: manifestation = "WEAK_OR_NORMAL" gap_reason = "NONE" identity = rows[0] outputs.append( { "case_id": CASE_ID, "action_id": ACTION_ID, "run_id": args.run_id, "company_market_id": identity["company_market_id"], "symbol": symbol, "company_id": identity["company_id"], "canonical_name": identity["canonical_name"], "universe_layer": identity["universe_layer"], "priority_bucket": identity["priority_bucket"], "price_rows": price_rows, "close_computable_days": close_days, "daily_return_computable_days": return_days, "amount_ratio_computable_days": amount_days, "first_trade_date": first["row"]["trade_date"] if first else "", "last_trade_date": last["row"]["trade_date"] if last else "", "first_close": decimal_text(first_close), "last_close": decimal_text(last_close), "period_return_pct": decimal_text(period_return), "strong_up_days": strong_up_days, "max_amount_ratio_20": decimal_text(max_amount_ratio), "metric_completeness_status": "COMPLETE" if complete else "INSUFFICIENT_DATA", "manifestation_type": manifestation, "gap_reason": gap_reason, "source_latest_trade_date": source_latest, "currentity_status": "SOURCE_CURRENTITY_GATE_PASSED_AS_OF_2026-07-24", "raw_snapshot_sha256": raw_sha256, "raw_snapshot_bytes": len(raw_bytes), "raw_snapshot_rows": len(raw_rows), "summary_derivation_version": "RAW_SNAPSHOT_LOCAL_DERIVATION_V1", "conclusion_strength": "MARKET_OBSERVATION_ONLY_NOT_INVESTMENT_CONCLUSION", "review_status": "PENDING_INDEPENDENT_EXECUTION_AND_OUTPUT_QUALITY_REVIEW", "formal_pool_effect": "NO_AUTOMATIC_FORMAL_POOL_CHANGE", } ) output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=OUTPUT_COLUMNS, lineterminator="\n") writer.writeheader() writer.writerows(outputs) return 0 if __name__ == "__main__": raise SystemExit(main())