#!/usr/bin/env python3 """Synthesize evidence-backed, OHLCV-only Wave 1 feature candidates. This is a read-only reporting tool. It does not evaluate formulas, search strategies, infer data provenance, or turn associative evidence into causality. """ from __future__ import annotations import argparse import hashlib import json from collections import defaultdict from pathlib import Path from typing import Any ARTIFACT = "HYPERSCALPER_WAVE1_FEATURE_SYNTHESIS_V1" MIN_CANDIDATES = 20 MAX_CANDIDATES = 50 ASSOCIATIVE_LIMITATION = ( "Evidence is observational and multi-feature attribution is associative, not causal." ) def load_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError(f"{path} must contain a JSON object") return value def source_record(path: Path | None) -> dict[str, Any]: if path is None: return {"status": "UNAVAILABLE", "reason": "not supplied"} return { "status": "AVAILABLE", "path": str(path), "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } def rows(value: dict[str, Any], *names: str) -> list[dict[str, Any]]: for name in names: found = value.get(name) if isinstance(found, list): return [item for item in found if isinstance(item, dict)] return [] def feature_key(row: dict[str, Any]) -> str | None: value = row.get("feature_key") if value is not None: return str(value) try: return f"{int(row['indicator_id'])}:{int(row['period'])}:{float(row['p1']):g}" except (KeyError, TypeError, ValueError): return None def is_ohlcv_only(row: dict[str, Any]) -> bool: """Accept only an explicit, finite OHLCV dependency declaration.""" scope = str(row.get("input_scope", row.get("data_scope", ""))).upper() if scope in {"OHLCV", "OHLCV_ONLY"}: return True dependencies = row.get("data_dependencies", row.get("inputs")) if not isinstance(dependencies, list) or not dependencies: return False return {str(item).lower() for item in dependencies} <= { "open", "high", "low", "close", "volume", } def numeric(value: Any) -> float: return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else 0.0 def failure_evidence(payload: dict[str, Any]) -> dict[str, list[dict[str, Any]]]: result: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in rows(payload, "feature_failure_comparisons", "features", "comparisons"): key = feature_key(row) if key: result[key].append(row) return result def deq_evidence(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: deq = payload.get("decision_equivalence", payload) if not isinstance(deq, dict): return {} return { str(row["feature_key"]): row for row in rows(deq, "feature_summaries", "features") if row.get("feature_key") is not None } def redundancy_evidence(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: result: dict[str, dict[str, Any]] = {} for cluster in rows(payload.get("redundancy", payload), "clusters"): for key in cluster.get("members", []): result[str(key)] = { "cluster_id": cluster.get("cluster_id"), "cluster_size": cluster.get("size"), "representative_feature_key": cluster.get("representative_feature_key"), } return result def synthesis( semantic: dict[str, Any], redundancy: dict[str, Any], failure: dict[str, Any] | None, deq: dict[str, Any], lineage_gap: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[str], int]: feature_rows = rows(semantic, "features", "primitives", "rows", "semantic_map") if not feature_rows: feature_rows = rows(lineage_gap.get("information_map", {}), "features") failures = failure_evidence(failure) if failure else {} deq_by_feature = deq_evidence(deq) clusters = redundancy_evidence(redundancy) lineage_by_feature = { str(row.get("feature_key")): row for row in rows(lineage_gap.get("information_map", {}), "features") if row.get("feature_key") is not None } excluded_non_ohlcv = 0 candidates = [] for row in feature_rows: key = feature_key(row) if key is None: continue if not is_ohlcv_only(row): excluded_non_ohlcv += 1 continue lineage = lineage_by_feature.get(key, row) feature_failures = failures.get(key, []) candidates.append( { "candidate_id": f"ohlcv:{key}", "feature_key": key, "indicator_id": row.get("indicator_id"), "period": row.get("period"), "p1": row.get("p1"), "domain": row.get("domain", row.get("semantic_domain", "unclassified")), "output_type": row.get("output_type", row.get("semantic_type", "unclassified")), "input_scope": "OHLCV_ONLY", "evidence": { "semantic": {"feature_key": key, "explicit_ohlcv_only": True}, "lineage_gap": { "historical_usage": lineage.get("historical_usage"), "lineage_mentions": lineage.get("lineage_mentions"), "max_redundancy": lineage.get("max_redundancy"), "redundancy_status": lineage.get("redundancy_status"), }, "redundancy_cluster": clusters.get(key), "failure_mining": feature_failures or None, "decision_equivalence": deq_by_feature.get(key), }, "limitations": [ ASSOCIATIVE_LIMITATION, "Not a strategy approval or causal feature claim.", ], } ) # Evidence ordering only: failure contrast, DEQ observations, lineage use, then redundancy. def rank(candidate: dict[str, Any]) -> tuple[float, float, float, float, str]: failure_rows = candidate["evidence"]["failure_mining"] or [] failure_strength = max( (numeric(item.get("ks_statistic")) for item in failure_rows), default=0.0 ) deq_row = candidate["evidence"]["decision_equivalence"] or {} deq_support = numeric(deq_row.get("trade_count")) lineage = candidate["evidence"]["lineage_gap"] usage = numeric(lineage.get("historical_usage")) + numeric(lineage.get("lineage_mentions")) redundancy_value = lineage.get("max_redundancy") redundancy = numeric(redundancy_value) if redundancy_value is not None else 1.0 return (-failure_strength, -deq_support, -usage, redundancy, candidate["feature_key"]) return sorted(candidates, key=rank), [], excluded_non_ohlcv def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--semantic-summary", type=Path, required=True) parser.add_argument("--redundancy-clusters", type=Path, required=True) parser.add_argument("--deq-summary", type=Path, required=True) parser.add_argument("--lineage-gap-analysis", type=Path, required=True) parser.add_argument( "--failure-mining", type=Path, help="Copied Spark failure-mining JSON when available" ) parser.add_argument("--output-dir", type=Path, required=True) args = parser.parse_args() required = ( args.semantic_summary, args.redundancy_clusters, args.deq_summary, args.lineage_gap_analysis, ) if any(not path.is_file() for path in required) or ( args.failure_mining and not args.failure_mining.is_file() ): parser.error("all supplied artifact paths must be files") semantic, redundancy, deq, lineage = (load_json(path) for path in required) failure = load_json(args.failure_mining) if args.failure_mining else None candidates, blockers, excluded_non_ohlcv = synthesis( semantic, redundancy, failure, deq, lineage ) selected = candidates[:MAX_CANDIDATES] if args.failure_mining is None: blockers.append( "Spark failure-mining result is unavailable; copy its JSON before relying on " "failure evidence." ) if len(selected) < MIN_CANDIDATES: blockers.append( f"Only {len(selected)} explicitly OHLCV-only candidates are evidenced; refusing to " f"invent the required {MIN_CANDIDATES}-{MAX_CANDIDATES} selection." ) status = "READY" if not blockers else "BLOCKED" sources = { "semantic_summary": source_record(args.semantic_summary), "redundancy_clusters": source_record(args.redundancy_clusters), "failure_mining": source_record(args.failure_mining), "deq_summary": source_record(args.deq_summary), "lineage_gap_analysis": source_record(args.lineage_gap_analysis), } common = { "schema_version": 1, "artifact": ARTIFACT, "read_only": True, "strategy_search": "not performed", "associative_limitations": ASSOCIATIVE_LIMITATION, "status": status, "sources": sources, "blockers": blockers, } taxonomy = { **common, "artifact": "FEATURE_GAP_TAXONOMY_V1", "candidates": selected, "excluded_without_explicit_ohlcv_only_declaration": excluded_non_ohlcv, } plan = { **common, "artifact": "FEATURE_EXPANSION_PLAN_V1", "wave": 1, "admission_rule": "Explicit OHLCV-only declaration and evidence-ranked source records.", "candidate_count": len(selected), "candidates": [item["candidate_id"] for item in selected], } registry = {**common, "artifact": "NEW_FEATURE_CANDIDATE_REGISTRY_V1", "candidates": selected} spec = { **common, "artifact": "HYPERSCALPER_WAVE1_FEATURE_SPEC_V1", "features": selected, "implementation_constraint": ( "Implement features from OHLCV inputs only; no strategy search or causal claims." ), } args.output_dir.mkdir(parents=True, exist_ok=True) outputs = { "feature_gap_taxonomy_v1.json": taxonomy, "feature_expansion_plan_v1.json": plan, "new_feature_candidate_registry_v1.json": registry, "hyperscalper_wave1_feature_spec_v1.json": spec, } for name, payload in outputs.items(): (args.output_dir / name).write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) markdown = ( "# HYPERSCALPER Wave 1 Feature Spec V1\n\n" f"Status: {status}. Read-only: yes. Strategy search: not performed.\n\n" f"Selected candidates: {len(selected)}. Explicitly excluded without OHLCV-only evidence: " f"{excluded_non_ohlcv}.\n\n" "## Limitations\n\n" f"- {ASSOCIATIVE_LIMITATION}\n" "- No feature is a strategy, approval, or causal claim.\n\n" "## Candidates\n\n" + "\n".join( f"- `{item['candidate_id']}`: {item['domain']} / {item['output_type']}" for item in selected ) + "\n\n## Blockers\n\n" + "\n".join(f"- {item}" for item in blockers) + "\n" ) (args.output_dir / "hyperscalper_wave1_feature_spec_v1.md").write_text( markdown, encoding="utf-8" ) executive = ( f"Wave 1 synthesis is {status}: {len(selected)} evidence-ranked, explicitly OHLCV-only " "candidates. " f"Failure mining is {'available' if failure else 'unavailable'}. {ASSOCIATIVE_LIMITATION}\n" ) (args.output_dir / "hyperscalper_wave1_executive_summary_v1.md").write_text( executive, encoding="utf-8" ) print(args.output_dir / "feature_gap_taxonomy_v1.json") if __name__ == "__main__": main()