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())
|
