331 lines
12 KiB
Python
331 lines
12 KiB
Python
|
"""Generate metadata-only engineering maps for the observed historical frontier.
|
||
|
|
|
||
|
|
The historical dispatcher is parsed as source text. No formula module is
|
||
|
|
imported, evaluated, copied, or otherwise ported by this generator.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import ast
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pyarrow as pa
|
||
|
|
import pyarrow.parquet as pq
|
||
|
|
|
||
|
|
FRONTIER_SIZE = 568
|
||
|
|
ROLES = ("trend", "signal", "trigger", "confirm", "vol")
|
||
|
|
FAMILY_PREFIXES = (
|
||
|
|
("ma_", "moving_average"),
|
||
|
|
("osc_", "oscillator"),
|
||
|
|
("vol_", "volatility"),
|
||
|
|
("regime_", "regime"),
|
||
|
|
("micro_", "microstructure"),
|
||
|
|
("mom_", "momentum"),
|
||
|
|
("trend_", "trend_structure"),
|
||
|
|
("norm_", "normalization"),
|
||
|
|
("cross_", "cross_indicator"),
|
||
|
|
("time_", "time_session"),
|
||
|
|
("pivot_", "reference_level"),
|
||
|
|
("prev_", "reference_level"),
|
||
|
|
("rolling_", "reference_level"),
|
||
|
|
("linreg_channel_", "reference_level"),
|
||
|
|
("quantile_", "reference_level"),
|
||
|
|
("donch_", "band_channel"),
|
||
|
|
("kelt_", "band_channel"),
|
||
|
|
("bb_", "band_channel"),
|
||
|
|
("ichimoku_", "band_channel"),
|
||
|
|
("supertrend", "band_channel"),
|
||
|
|
("psar", "band_channel"),
|
||
|
|
("adx", "trend_structure"),
|
||
|
|
("aroon_", "trend_structure"),
|
||
|
|
("linreg_", "trend_structure"),
|
||
|
|
("entropy_", "regime"),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def digest(path: Path) -> str:
|
||
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def canonical_json(payload: object) -> bytes:
|
||
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False).encode(
|
||
|
|
"ascii"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def function_calls(node: ast.AST, functions: set[str]) -> list[str]:
|
||
|
|
return [
|
||
|
|
child.func.id
|
||
|
|
for child in ast.walk(node)
|
||
|
|
if (
|
||
|
|
isinstance(child, ast.Call)
|
||
|
|
and isinstance(child.func, ast.Name)
|
||
|
|
and child.func.id in functions
|
||
|
|
)
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def source_map(path: Path) -> dict[int, dict[str, object]]:
|
||
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
||
|
|
functions = {node.name: node for node in tree.body if isinstance(node, ast.FunctionDef)}
|
||
|
|
dispatcher = functions.get("compute_indicator")
|
||
|
|
if dispatcher is None:
|
||
|
|
raise ValueError("historical source has no compute_indicator dispatcher")
|
||
|
|
mapping: dict[int, dict[str, object]] = {}
|
||
|
|
for node in ast.walk(dispatcher):
|
||
|
|
if not isinstance(node, ast.Compare) or len(node.ops) != 1:
|
||
|
|
continue
|
||
|
|
if not (
|
||
|
|
isinstance(node.left, ast.Name)
|
||
|
|
and node.left.id == "ind_id"
|
||
|
|
and isinstance(node.ops[0], ast.Eq)
|
||
|
|
and len(node.comparators) == 1
|
||
|
|
and isinstance(node.comparators[0], ast.Constant)
|
||
|
|
and isinstance(node.comparators[0].value, int)
|
||
|
|
):
|
||
|
|
continue
|
||
|
|
parent = next(
|
||
|
|
(
|
||
|
|
candidate
|
||
|
|
for candidate in ast.walk(dispatcher)
|
||
|
|
if isinstance(candidate, ast.If) and candidate.test is node
|
||
|
|
),
|
||
|
|
None,
|
||
|
|
)
|
||
|
|
if parent is None:
|
||
|
|
continue
|
||
|
|
returned = next((item.value for item in parent.body if isinstance(item, ast.Return)), None)
|
||
|
|
if returned is None:
|
||
|
|
continue
|
||
|
|
calls = function_calls(returned, set(functions))
|
||
|
|
primary = calls[0] if calls else None
|
||
|
|
if primary is None:
|
||
|
|
continue
|
||
|
|
signature = functions[primary].args
|
||
|
|
inputs = [
|
||
|
|
argument.arg
|
||
|
|
for argument in signature.args
|
||
|
|
if argument.arg in {"close", "high", "low", "volume"}
|
||
|
|
]
|
||
|
|
dependencies = sorted(set(function_calls(functions[primary], set(functions))) - {primary})
|
||
|
|
mapping[node.comparators[0].value] = {
|
||
|
|
"function": primary,
|
||
|
|
"source_expression": ast.unparse(returned),
|
||
|
|
"required_inputs": inputs,
|
||
|
|
"helper_dependencies": dependencies,
|
||
|
|
}
|
||
|
|
return mapping
|
||
|
|
|
||
|
|
|
||
|
|
def family(function: str) -> str:
|
||
|
|
for prefix, value in FAMILY_PREFIXES:
|
||
|
|
if function.startswith(prefix):
|
||
|
|
return value
|
||
|
|
return "specialized"
|
||
|
|
|
||
|
|
|
||
|
|
def counts(rows: list[dict[str, object]]) -> dict[str, int]:
|
||
|
|
return {
|
||
|
|
"primitives": len(rows),
|
||
|
|
"indicator_ids": len({int(row["indicator_id"]) for row in rows}),
|
||
|
|
"functions": len({str(row["source_function"]) for row in rows}),
|
||
|
|
"families": len({str(row["engineering_family"]) for row in rows}),
|
||
|
|
"parameter_expansions": len(rows) - len({int(row["indicator_id"]) for row in rows}),
|
||
|
|
"role_duplicates": sum(int(row["role_triple_count"]) - 1 for row in rows),
|
||
|
|
"native_reuse_primitives": sum(
|
||
|
|
row["implementation_path"] == "native_reuse" for row in rows
|
||
|
|
),
|
||
|
|
"native_reuse_functions": len(
|
||
|
|
{
|
||
|
|
str(row["source_function"])
|
||
|
|
for row in rows
|
||
|
|
if row["implementation_path"] == "native_reuse"
|
||
|
|
}
|
||
|
|
),
|
||
|
|
"new_algorithm_primitives": sum(
|
||
|
|
row["implementation_path"] == "new_algorithm" for row in rows
|
||
|
|
),
|
||
|
|
"new_algorithm_functions": len(
|
||
|
|
{
|
||
|
|
str(row["source_function"])
|
||
|
|
for row in rows
|
||
|
|
if row["implementation_path"] == "new_algorithm"
|
||
|
|
}
|
||
|
|
),
|
||
|
|
"alias_refused_primitives": sum(
|
||
|
|
row["implementation_path"] == "alias_refused" for row in rows
|
||
|
|
),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--coverage", type=Path, required=True)
|
||
|
|
parser.add_argument("--registry", type=Path, required=True)
|
||
|
|
parser.add_argument("--historical-source", type=Path, required=True)
|
||
|
|
parser.add_argument("--output-dir", type=Path, required=True)
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
coverage = json.loads(args.coverage.read_text(encoding="utf-8"))
|
||
|
|
registry = json.loads(args.registry.read_text(encoding="utf-8"))
|
||
|
|
source = source_map(args.historical_source)
|
||
|
|
definitions = {item["indicator_id"]: item for item in registry["definitions"]}
|
||
|
|
rows: list[dict[str, object]] = []
|
||
|
|
for rank, primitive in enumerate(coverage["canonical_primitives"], start=1):
|
||
|
|
indicator_id = int(primitive["indicator_id"])
|
||
|
|
source_item = source.get(indicator_id)
|
||
|
|
if source_item is None:
|
||
|
|
raise ValueError(
|
||
|
|
f"observed indicator {indicator_id} is absent from the historical dispatcher"
|
||
|
|
)
|
||
|
|
definition = definitions[indicator_id]
|
||
|
|
source_function = str(source_item["function"])
|
||
|
|
rows.append(
|
||
|
|
{
|
||
|
|
"weighted_rank": rank,
|
||
|
|
"frontier": "95" if rank <= FRONTIER_SIZE else "95_to_100",
|
||
|
|
**primitive,
|
||
|
|
"registry_id": indicator_id,
|
||
|
|
"registry_name": definition["name"],
|
||
|
|
"registry_alias_of": definition["alias_of"],
|
||
|
|
"registry_behavior": definition["behavior"],
|
||
|
|
"source_function": source_function,
|
||
|
|
"source_expression": source_item["source_expression"],
|
||
|
|
"engineering_family": family(source_function),
|
||
|
|
"parameter_aliases": {"period": "period", "p1": "p1"},
|
||
|
|
"required_inputs": source_item["required_inputs"],
|
||
|
|
"helper_dependencies": source_item["helper_dependencies"],
|
||
|
|
"implementation_path": (
|
||
|
|
"alias_refused" if definition["status"] == "alias" else "unclassified"
|
||
|
|
),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
if len(rows) != 711 or len(rows[:FRONTIER_SIZE]) != FRONTIER_SIZE:
|
||
|
|
raise ValueError("expected exactly 711 observed primitives and a 568-item 95% frontier")
|
||
|
|
native_functions = {
|
||
|
|
str(row["source_function"]) for row in rows if row["coverage_classification"] == "native"
|
||
|
|
}
|
||
|
|
for row in rows:
|
||
|
|
if row["implementation_path"] == "alias_refused":
|
||
|
|
continue
|
||
|
|
row["implementation_path"] = (
|
||
|
|
"native_reuse" if row["source_function"] in native_functions else "new_algorithm"
|
||
|
|
)
|
||
|
|
|
||
|
|
selected, delta = rows[:FRONTIER_SIZE], rows[FRONTIER_SIZE:]
|
||
|
|
total_slots = sum(int(row["usage_slots"]) for row in rows)
|
||
|
|
selected_slots = sum(int(row["usage_slots"]) for row in selected)
|
||
|
|
leverage = {
|
||
|
|
"slots_per_new_algorithm_function": (
|
||
|
|
selected_slots / counts(selected)["new_algorithm_functions"]
|
||
|
|
),
|
||
|
|
"primitives_per_new_algorithm_function": (
|
||
|
|
counts(selected)["new_algorithm_primitives"]
|
||
|
|
/ counts(selected)["new_algorithm_functions"]
|
||
|
|
),
|
||
|
|
"parameter_expansions_per_function": (
|
||
|
|
counts(selected)["parameter_expansions"] / counts(selected)["functions"]
|
||
|
|
),
|
||
|
|
"role_assignments_per_primitive": (
|
||
|
|
sum(int(row["role_triple_count"]) for row in selected) / len(selected)
|
||
|
|
),
|
||
|
|
}
|
||
|
|
family_rows = []
|
||
|
|
for name in sorted({str(row["engineering_family"]) for row in rows}):
|
||
|
|
items = [row for row in rows if row["engineering_family"] == name]
|
||
|
|
family_rows.append(
|
||
|
|
{
|
||
|
|
"engineering_family": name,
|
||
|
|
"counts": counts(items),
|
||
|
|
"usage_slots": sum(int(row["usage_slots"]) for row in items),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
source_artifacts = {
|
||
|
|
"coverage": {"path": str(args.coverage), "sha256": digest(args.coverage)},
|
||
|
|
"registry": {"path": str(args.registry), "sha256": digest(args.registry)},
|
||
|
|
"historical_source": {
|
||
|
|
"path": str(args.historical_source),
|
||
|
|
"sha256": digest(args.historical_source),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
family_payload = {
|
||
|
|
"schema_version": 1,
|
||
|
|
"artifact": "ENGINEERING_FAMILY_MAP_V1",
|
||
|
|
"method": (
|
||
|
|
"static local archaeology only; historical dispatcher AST and registry metadata; "
|
||
|
|
"no formula import, evaluation, or port"
|
||
|
|
),
|
||
|
|
"source_artifacts": source_artifacts,
|
||
|
|
"counts": counts(rows),
|
||
|
|
"families": family_rows,
|
||
|
|
"primitives": rows,
|
||
|
|
}
|
||
|
|
frontier_payload = {
|
||
|
|
"schema_version": 1,
|
||
|
|
"artifact": "IMPLEMENTATION_FRONTIER_V1",
|
||
|
|
"method": family_payload["method"],
|
||
|
|
"source_artifacts": source_artifacts,
|
||
|
|
"frontier_95": {
|
||
|
|
"primitive_count": len(selected),
|
||
|
|
"usage_slots": selected_slots,
|
||
|
|
"coverage_percent": selected_slots * 100 / total_slots,
|
||
|
|
"counts": counts(selected),
|
||
|
|
"leverage": leverage,
|
||
|
|
},
|
||
|
|
"full_100": {
|
||
|
|
"primitive_count": len(rows),
|
||
|
|
"usage_slots": total_slots,
|
||
|
|
"coverage_percent": 100.0,
|
||
|
|
"counts": counts(rows),
|
||
|
|
},
|
||
|
|
"delta_95_to_100": {
|
||
|
|
"primitive_count": len(delta),
|
||
|
|
"usage_slots": sum(int(row["usage_slots"]) for row in delta),
|
||
|
|
"coverage_percent": (total_slots - selected_slots) * 100 / total_slots,
|
||
|
|
"incremental_counts": counts(delta),
|
||
|
|
},
|
||
|
|
"frontier_primitives": selected,
|
||
|
|
"delta_primitives": delta,
|
||
|
|
}
|
||
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
(args.output_dir / "engineering_family_map_v1.json").write_bytes(
|
||
|
|
canonical_json(family_payload) + b"\n"
|
||
|
|
)
|
||
|
|
(args.output_dir / "implementation_frontier_v1.json").write_bytes(
|
||
|
|
canonical_json(frontier_payload) + b"\n"
|
||
|
|
)
|
||
|
|
parquet_rows = [
|
||
|
|
{
|
||
|
|
**{
|
||
|
|
key: value
|
||
|
|
for key, value in row.items()
|
||
|
|
if key
|
||
|
|
not in {
|
||
|
|
"roles",
|
||
|
|
"role_triples",
|
||
|
|
"parameter_aliases",
|
||
|
|
"required_inputs",
|
||
|
|
"helper_dependencies",
|
||
|
|
}
|
||
|
|
},
|
||
|
|
"roles_json": json.dumps(row["roles"], sort_keys=True),
|
||
|
|
"role_triples_json": json.dumps(row["role_triples"], sort_keys=True),
|
||
|
|
"parameter_aliases_json": json.dumps(row["parameter_aliases"], sort_keys=True),
|
||
|
|
"required_inputs_json": json.dumps(row["required_inputs"]),
|
||
|
|
"helper_dependencies_json": json.dumps(row["helper_dependencies"]),
|
||
|
|
}
|
||
|
|
for row in rows
|
||
|
|
]
|
||
|
|
table = pa.Table.from_pylist(parquet_rows)
|
||
|
|
pq.write_table(table, args.output_dir / "engineering_family_map_v1.parquet", compression="zstd")
|
||
|
|
pq.write_table(
|
||
|
|
table, args.output_dir / "implementation_frontier_v1.parquet", compression="zstd"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|