import csv
|
import hashlib
|
import json
|
from pathlib import Path
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
RAW_DIR = PROJECT_ROOT / "exp-data" / "raw"
|
RESULT_DIR = PROJECT_ROOT / "exp-data" / "result"
|
IMG_DIR = PROJECT_ROOT / "exp-data" / "img"
|
|
|
def sha256_file(path: Path) -> str:
|
h = hashlib.sha256()
|
with path.open("rb") as f:
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
h.update(chunk)
|
return h.hexdigest()
|
|
|
def write_csv(path: Path, rows: list[dict], fieldnames: list[str]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", newline="", encoding="utf-8") as f:
|
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def write_json(path: Path, payload: dict) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
def write_readout(path: Path, title: str, lines: list[str]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
body = [f"# {title}", ""]
|
body.extend(f"- {line}" for line in lines)
|
path.write_text("\n".join(body) + "\n", encoding="utf-8")
|
|
|
def write_svg_line_chart(path: Path, rows: list[dict], value_field: str, signal_field: str, title: str) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
width, height = 720, 320
|
margin = 42
|
values = [float(r[value_field]) for r in rows]
|
min_v, max_v = min(values), max(values)
|
span = max(max_v - min_v, 1.0)
|
|
def xy(i: int, value: float) -> tuple[float, float]:
|
x = margin + i * ((width - margin * 2) / max(len(rows) - 1, 1))
|
y = height - margin - ((value - min_v) / span) * (height - margin * 2)
|
return x, y
|
|
points = [xy(i, v) for i, v in enumerate(values)]
|
poly = " ".join(f"{x:.1f},{y:.1f}" for x, y in points)
|
circles = []
|
for i, r in enumerate(rows):
|
if str(r.get(signal_field, "0")) == "1":
|
x, y = xy(i, float(r[value_field]))
|
circles.append(f'<circle cx="{x:.1f}" cy="{y:.1f}" r="6" fill="#d62828" />')
|
circles.append(f'<text x="{x + 8:.1f}" y="{y - 8:.1f}" font-size="12" fill="#d62828">signal</text>')
|
svg = f"""<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" viewBox="0 0 {width} {height}">
|
<rect width="100%" height="100%" fill="#fffaf0"/>
|
<text x="{margin}" y="24" font-size="18" font-family="sans-serif" fill="#222">{title}</text>
|
<line x1="{margin}" y1="{height-margin}" x2="{width-margin}" y2="{height-margin}" stroke="#444"/>
|
<line x1="{margin}" y1="{margin}" x2="{margin}" y2="{height-margin}" stroke="#444"/>
|
<polyline fill="none" stroke="#1f77b4" stroke-width="3" points="{poly}"/>
|
{''.join(circles)}
|
</svg>
|
"""
|
path.write_text(svg, encoding="utf-8")
|
|
|
def build_manifest(run_dir: Path, artifacts: list[tuple[str, Path, str, int]]) -> None:
|
rows = []
|
for name, path, role, count in artifacts:
|
rows.append(
|
{
|
"artifact": name,
|
"path": path.relative_to(PROJECT_ROOT).as_posix(),
|
"artifact_role": role,
|
"row_count_or_count": count,
|
"sha256": sha256_file(path),
|
}
|
)
|
write_csv(
|
run_dir / "output_manifest.csv",
|
rows,
|
["artifact", "path", "artifact_role", "row_count_or_count", "sha256"],
|
)
|
|
|
def run_signal_cross() -> None:
|
experiment_id = "EXP-20260601-PROJA-SIGNAL-CROSS-001"
|
design_id = "DESIGN-20260601-PROJA-SIGNAL-CROSS-001"
|
run_id = "RUN-20260601-PROJA-SIGNAL-CROSS-001"
|
run_dir = RESULT_DIR / run_id
|
intermediate_dir = run_dir / "intermediate"
|
img_dir = IMG_DIR / experiment_id / run_id
|
|
raw_rows = [
|
{"experiment_id": experiment_id, "object_id": "OBJECT001", "event_date": f"2026-01-{i+1:02d}", "value": v}
|
for i, v in enumerate([10, 11, 12, 14, 13, 15, 18, 16])
|
]
|
raw_path = RAW_DIR / run_id / "signal_cross_input.csv"
|
write_csv(raw_path, raw_rows, ["experiment_id", "object_id", "event_date", "value"])
|
|
feature_rows = []
|
signal_rows = []
|
for i, row in enumerate(raw_rows):
|
trailing = [float(r["value"]) for r in raw_rows[max(0, i - 3) : i]]
|
trailing_ma3 = sum(trailing) / len(trailing) if trailing else None
|
value = float(row["value"])
|
signal = 1 if trailing_ma3 is not None and value > trailing_ma3 * 1.08 else 0
|
out = dict(row)
|
out["trailing_ma3"] = "" if trailing_ma3 is None else f"{trailing_ma3:.4f}"
|
out["signal_flag"] = signal
|
feature_rows.append(out)
|
if signal:
|
signal_rows.append(
|
{
|
"experiment_id": experiment_id,
|
"design_id": design_id,
|
"run_id": run_id,
|
"object_id": row["object_id"],
|
"event_date": row["event_date"],
|
"value": row["value"],
|
"trailing_ma3": out["trailing_ma3"],
|
"signal_reason": "value_gt_trailing_ma3_by_8pct",
|
}
|
)
|
feature_path = intermediate_dir / "feature_panel.csv"
|
result_path = run_dir / "signal_cross_result.csv"
|
summary_path = run_dir / "summary.json"
|
readout_path = run_dir / "readout.md"
|
svg_path = img_dir / "OBJECT001_signal_cross.svg"
|
input_manifest_path = run_dir / "input_manifest.csv"
|
|
write_csv(feature_path, feature_rows, ["experiment_id", "object_id", "event_date", "value", "trailing_ma3", "signal_flag"])
|
write_csv(result_path, signal_rows, ["experiment_id", "design_id", "run_id", "object_id", "event_date", "value", "trailing_ma3", "signal_reason"])
|
write_svg_line_chart(svg_path, feature_rows, "value", "signal_flag", "Project A signal cross smoke")
|
write_json(
|
summary_path,
|
{
|
"experiment_id": experiment_id,
|
"design_id": design_id,
|
"run_id": run_id,
|
"status": "PASS",
|
"raw_row_count": len(raw_rows),
|
"feature_row_count": len(feature_rows),
|
"signal_count": len(signal_rows),
|
"future_function_required": False,
|
"read_as": "environment and evidence-chain smoke only",
|
},
|
)
|
write_readout(
|
readout_path,
|
"Project A signal cross smoke",
|
[
|
f"raw rows: {len(raw_rows)}",
|
f"feature rows: {len(feature_rows)}",
|
f"signal count: {len(signal_rows)}",
|
"This is synthetic data and only validates experiment-system wiring.",
|
],
|
)
|
write_csv(
|
input_manifest_path,
|
[{"artifact": "raw_input", "path": raw_path.relative_to(PROJECT_ROOT).as_posix(), "sha256": sha256_file(raw_path)}],
|
["artifact", "path", "sha256"],
|
)
|
build_manifest(
|
run_dir,
|
[
|
("raw_input", raw_path, "raw", len(raw_rows)),
|
("feature_panel", feature_path, "intermediate", len(feature_rows)),
|
("signal_cross_result", result_path, "output", len(signal_rows)),
|
("summary", summary_path, "summary", 1),
|
("readout", readout_path, "readout", 1),
|
("signal_cross_svg", svg_path, "image", 1),
|
("input_manifest", input_manifest_path, "manifest", 1),
|
],
|
)
|
|
|
def run_volume_spike() -> None:
|
experiment_id = "EXP-20260601-PROJA-VOLUME-SPIKE-002"
|
design_id = "DESIGN-20260601-PROJA-VOLUME-SPIKE-002"
|
run_id = "RUN-20260601-PROJA-VOLUME-SPIKE-002"
|
run_dir = RESULT_DIR / run_id
|
intermediate_dir = run_dir / "intermediate"
|
img_dir = IMG_DIR / experiment_id / run_id
|
|
values = [20, 19, 20, 21, 22, 21, 24, 25]
|
volumes = [100, 105, 98, 110, 260, 130, 300, 145]
|
raw_rows = [
|
{
|
"experiment_id": experiment_id,
|
"object_id": "OBJECT002",
|
"event_date": f"2026-02-{i+1:02d}",
|
"value": values[i],
|
"volume": volumes[i],
|
}
|
for i in range(len(values))
|
]
|
raw_path = RAW_DIR / run_id / "volume_spike_input.csv"
|
write_csv(raw_path, raw_rows, ["experiment_id", "object_id", "event_date", "value", "volume"])
|
|
feature_rows = []
|
signal_rows = []
|
for i, row in enumerate(raw_rows):
|
trailing = [float(r["volume"]) for r in raw_rows[max(0, i - 3) : i]]
|
volume_ma3 = sum(trailing) / len(trailing) if trailing else None
|
prev_value = float(raw_rows[i - 1]["value"]) if i > 0 else None
|
value = float(row["value"])
|
volume = float(row["volume"])
|
value_up = prev_value is not None and value > prev_value
|
signal = 1 if volume_ma3 is not None and volume > volume_ma3 * 1.8 and value_up else 0
|
out = dict(row)
|
out["volume_ma3"] = "" if volume_ma3 is None else f"{volume_ma3:.4f}"
|
out["value_up_flag"] = int(value_up)
|
out["signal_flag"] = signal
|
feature_rows.append(out)
|
if signal:
|
signal_rows.append(
|
{
|
"experiment_id": experiment_id,
|
"design_id": design_id,
|
"run_id": run_id,
|
"object_id": row["object_id"],
|
"event_date": row["event_date"],
|
"value": row["value"],
|
"volume": row["volume"],
|
"volume_ma3": out["volume_ma3"],
|
"signal_reason": "volume_gt_trailing_ma3_by_80pct_and_value_up",
|
}
|
)
|
feature_path = intermediate_dir / "feature_panel.csv"
|
result_path = run_dir / "volume_spike_result.csv"
|
summary_path = run_dir / "summary.json"
|
readout_path = run_dir / "readout.md"
|
svg_path = img_dir / "OBJECT002_volume_spike.svg"
|
input_manifest_path = run_dir / "input_manifest.csv"
|
|
write_csv(feature_path, feature_rows, ["experiment_id", "object_id", "event_date", "value", "volume", "volume_ma3", "value_up_flag", "signal_flag"])
|
write_csv(result_path, signal_rows, ["experiment_id", "design_id", "run_id", "object_id", "event_date", "value", "volume", "volume_ma3", "signal_reason"])
|
write_svg_line_chart(svg_path, feature_rows, "value", "signal_flag", "Project A volume spike smoke")
|
write_json(
|
summary_path,
|
{
|
"experiment_id": experiment_id,
|
"design_id": design_id,
|
"run_id": run_id,
|
"status": "PASS",
|
"raw_row_count": len(raw_rows),
|
"feature_row_count": len(feature_rows),
|
"signal_count": len(signal_rows),
|
"future_function_required": False,
|
"read_as": "environment and evidence-chain smoke only",
|
},
|
)
|
write_readout(
|
readout_path,
|
"Project A volume spike smoke",
|
[
|
f"raw rows: {len(raw_rows)}",
|
f"feature rows: {len(feature_rows)}",
|
f"signal count: {len(signal_rows)}",
|
"This is synthetic data and only validates experiment-system wiring.",
|
],
|
)
|
write_csv(
|
input_manifest_path,
|
[{"artifact": "raw_input", "path": raw_path.relative_to(PROJECT_ROOT).as_posix(), "sha256": sha256_file(raw_path)}],
|
["artifact", "path", "sha256"],
|
)
|
build_manifest(
|
run_dir,
|
[
|
("raw_input", raw_path, "raw", len(raw_rows)),
|
("feature_panel", feature_path, "intermediate", len(feature_rows)),
|
("volume_spike_result", result_path, "output", len(signal_rows)),
|
("summary", summary_path, "summary", 1),
|
("readout", readout_path, "readout", 1),
|
("volume_spike_svg", svg_path, "image", 1),
|
("input_manifest", input_manifest_path, "manifest", 1),
|
],
|
)
|
|
|
def main() -> None:
|
run_signal_cross()
|
run_volume_spike()
|
print("Project A experiments generated: 2")
|
|
|
if __name__ == "__main__":
|
main()
|
