from __future__ import annotations
import csv
import hashlib
import json
from datetime import date, timedelta
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[2]
def _rel(path: Path) -> str:
return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.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-sig") as handle:
writer = csv.DictWriter(handle, 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 _business_days(start: date, count: int) -> list[date]:
days: list[date] = []
cursor = start
while len(days) < count:
if cursor.weekday() < 5:
days.append(cursor)
cursor += timedelta(days=1)
return days
def _build_kline_rows(
*,
experiment_id: str,
run_id: str,
object_id: str,
closes: list[float],
volumes: list[int],
) -> list[dict]:
rows: list[dict] = []
for idx, (event_date, close, volume) in enumerate(
zip(_business_days(date(2026, 5, 1), len(closes)), closes, volumes),
start=1,
):
prev_close = closes[idx - 2] if idx > 1 else close
open_price = round(prev_close * (1 + (0.002 if close >= prev_close else -0.002)), 2)
high = round(max(open_price, close) * 1.018, 2)
low = round(min(open_price, close) * 0.982, 2)
rows.append(
{
"experiment_id": experiment_id,
"run_id": run_id,
"object_id": object_id,
"event_date": event_date.isoformat(),
"open": f"{open_price:.2f}",
"high": f"{high:.2f}",
"low": f"{low:.2f}",
"close": f"{close:.2f}",
"volume": str(volume),
}
)
return rows
def _moving_average(values: list[float], window: int, idx: int) -> float | None:
if idx + 1 < window:
return None
return sum(values[idx + 1 - window : idx + 1]) / window
def _calculate_ma5_breakout(rows: list[dict]) -> list[dict]:
closes = [float(row["close"]) for row in rows]
result: list[dict] = []
for idx, row in enumerate(rows):
ma5 = _moving_average(closes, 5, idx)
prev_ma5 = _moving_average(closes, 5, idx - 1) if idx > 0 else None
prev_close = closes[idx - 1] if idx > 0 else None
flag = int(
ma5 is not None
and prev_ma5 is not None
and prev_close is not None
and prev_close <= prev_ma5
and closes[idx] > ma5
)
enriched = dict(row)
enriched.update(
{
"ma5": "" if ma5 is None else f"{ma5:.4f}",
"ma10": "",
"signal_type": "MA5_BREAKOUT" if flag else "",
"signal_flag": str(flag),
"signal_reason": "close_cross_above_ma5" if flag else "",
}
)
result.append(enriched)
return result
def _calculate_volume_pullback(rows: list[dict]) -> list[dict]:
closes = [float(row["close"]) for row in rows]
lows = [float(row["low"]) for row in rows]
volumes = [int(row["volume"]) for row in rows]
result: list[dict] = []
for idx, row in enumerate(rows):
ma10 = _moving_average(closes, 10, idx)
prev_volume = volumes[idx - 1] if idx > 0 else None
flag = int(
ma10 is not None
and prev_volume is not None
and lows[idx] <= ma10 * 1.015
and closes[idx] >= ma10
and volumes[idx] < prev_volume
)
enriched = dict(row)
enriched.update(
{
"ma5": "",
"ma10": "" if ma10 is None else f"{ma10:.4f}",
"signal_type": "VOLUME_PULLBACK_SUPPORT" if flag else "",
"signal_flag": str(flag),
"signal_reason": "low_near_ma10_and_volume_contracts" if flag else "",
}
)
result.append(enriched)
return result
def _write_svg(path: Path, rows: list[dict], title: str, ma_field: str) -> None:
width = 980
height = 420
margin = 46
closes = [float(row["close"]) for row in rows]
highs = [float(row["high"]) for row in rows]
lows = [float(row["low"]) for row in rows]
prices = highs + lows
p_min = min(prices) * 0.985
p_max = max(prices) * 1.015
step = (width - margin * 2) / max(1, len(rows) - 1)
def x(idx: int) -> float:
return margin + step * idx
def y(price: float) -> float:
return height - margin - (price - p_min) / (p_max - p_min) * (height - margin * 2)
candle_parts: list[str] = []
ma_points: list[str] = []
signal_parts: list[str] = []
for idx, row in enumerate(rows):
open_price = float(row["open"])
close = float(row["close"])
high = float(row["high"])
low = float(row["low"])
color = "#c43c35" if close >= open_price else "#2477b3"
cx = x(idx)
body_y = min(y(open_price), y(close))
body_h = max(2, abs(y(open_price) - y(close)))
candle_parts.append(
f''
)
candle_parts.append(
f''
)
ma_value = row.get(ma_field, "")
if ma_value:
ma_points.append(f"{cx:.1f},{y(float(ma_value)):.1f}")
if row.get("signal_flag") == "1":
signal_parts.append(
f''
)
signal_parts.append(
f'signal'
)
ma_polyline = ""
if len(ma_points) >= 2:
ma_polyline = (
f''
)
svg = f"""
"""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(svg, encoding="utf-8")
def _csv_row_count(path: Path) -> int:
if path.suffix.lower() != ".csv":
return 1
with path.open("r", encoding="utf-8-sig", newline="") as handle:
reader = csv.reader(handle)
rows = list(reader)
return max(0, len(rows) - 1)
def _manifest_rows(paths: list[Path], role_map: dict[str, str]) -> list[dict]:
rows: list[dict] = []
for path in paths:
rows.append(
{
"artifact": path.name,
"path": _rel(path),
"artifact_role": role_map.get(path.name, "artifact"),
"row_count_or_count": str(_csv_row_count(path)),
"sha256": _sha256(path),
}
)
return rows
def _run_one(config: dict) -> dict:
run_id = config["run_id"]
experiment_id = config["experiment_id"]
raw_dir = PROJECT_ROOT / "exp-data" / "raw" / run_id
result_dir = PROJECT_ROOT / "exp-data" / "result" / run_id
img_dir = PROJECT_ROOT / "exp-data" / "img" / experiment_id / run_id
raw_path = raw_dir / "synthetic_kline_input.csv"
intermediate_path = result_dir / "intermediate" / "kline_feature_panel.csv"
result_path = result_dir / config["result_file"]
svg_path = img_dir / config["chart_file"]
readout_path = result_dir / "readout.md"
summary_path = result_dir / "summary.json"
input_manifest_path = result_dir / "input_manifest.csv"
output_manifest_path = result_dir / "output_manifest.csv"
raw_rows = _build_kline_rows(
experiment_id=experiment_id,
run_id=run_id,
object_id=config["object_id"],
closes=config["closes"],
volumes=config["volumes"],
)
_write_csv(raw_path, raw_rows, ["experiment_id", "run_id", "object_id", "event_date", "open", "high", "low", "close", "volume"])
if config["kind"] == "ma5_breakout":
result_rows = _calculate_ma5_breakout(raw_rows)
ma_field = "ma5"
elif config["kind"] == "volume_pullback":
result_rows = _calculate_volume_pullback(raw_rows)
ma_field = "ma10"
else:
raise ValueError(config["kind"])
feature_fields = [
"experiment_id",
"run_id",
"object_id",
"event_date",
"close",
"volume",
"ma5",
"ma10",
]
feature_rows = [{field: row.get(field, "") for field in feature_fields} for row in result_rows]
_write_csv(intermediate_path, feature_rows, feature_fields)
result_fields = [
"experiment_id",
"run_id",
"object_id",
"event_date",
"open",
"high",
"low",
"close",
"volume",
"ma5",
"ma10",
"signal_type",
"signal_flag",
"signal_reason",
]
_write_csv(result_path, result_rows, result_fields)
_write_svg(svg_path, result_rows, config["title"], ma_field)
signal_count = sum(int(row["signal_flag"]) for row in result_rows)
summary = {
"experiment_id": experiment_id,
"design_id": config["design_id"],
"run_id": run_id,
"status": "PASS" if signal_count >= 1 else "FAIL",
"data_type": "synthetic_kline",
"result_boundary": "environment_and_recording_flow_validation_only",
"row_count": len(result_rows),
"object_count": 1,
"signal_count": signal_count,
"raw_input": _rel(raw_path),
"intermediate_feature_panel": _rel(intermediate_path),
"result_table": _rel(result_path),
"chart": _rel(svg_path),
}
_write_json(summary_path, summary)
readout = f"""# {experiment_id} readout
- status: {summary["status"]}
- design_id: {config["design_id"]}
- run_id: {run_id}
- data_type: synthetic_kline
- row_count: {summary["row_count"]}
- signal_count: {signal_count}
- intermediate_feature_panel: `{_rel(intermediate_path)}`
- result_table: `{_rel(result_path)}`
- chart: `{_rel(svg_path)}`
Boundary: this experiment validates project B experiment recording and artifact chain only. It does not prove any real-world trading rule.
"""
readout_path.write_text(readout, encoding="utf-8")
_write_csv(
input_manifest_path,
_manifest_rows([raw_path], {raw_path.name: "raw_input"}),
["artifact", "path", "artifact_role", "row_count_or_count", "sha256"],
)
_write_csv(
output_manifest_path,
_manifest_rows(
[intermediate_path, result_path, svg_path, summary_path, readout_path, input_manifest_path],
{
result_path.name: "result_table",
intermediate_path.name: "intermediate_feature_panel",
svg_path.name: "chart",
summary_path.name: "summary",
readout_path.name: "readout",
input_manifest_path.name: "input_manifest",
},
),
["artifact", "path", "artifact_role", "row_count_or_count", "sha256"],
)
return summary
def main() -> None:
configs = [
{
"experiment_id": "EXP-20260601-KLINE-MA5-BREAKOUT-001",
"design_id": "DESIGN-20260601-KLINE-MA5-BREAKOUT-001",
"run_id": "RUN-20260601-KLINE-MA5-BREAKOUT-001",
"object_id": "KOBJ001",
"kind": "ma5_breakout",
"title": "KOBJ001 synthetic K-line MA5 breakout",
"result_file": "kline_ma5_breakout_result.csv",
"chart_file": "KOBJ001_ma5_breakout.svg",
"closes": [
10.00,
9.90,
9.82,
9.78,
9.74,
9.70,
9.72,
9.75,
9.78,
10.05,
10.28,
10.55,
10.46,
10.62,
10.82,
10.76,
10.94,
11.10,
11.05,
11.22,
11.35,
11.28,
11.42,
11.58,
11.66,
],
"volumes": [
2100,
1980,
1880,
1820,
1760,
1690,
1710,
1770,
1840,
2450,
2680,
2920,
2510,
2640,
2770,
2380,
2460,
2580,
2300,
2420,
2550,
2260,
2360,
2490,
2520,
],
},
{
"experiment_id": "EXP-20260601-KLINE-VOLUME-PULLBACK-002",
"design_id": "DESIGN-20260601-KLINE-VOLUME-PULLBACK-002",
"run_id": "RUN-20260601-KLINE-VOLUME-PULLBACK-002",
"object_id": "KOBJ002",
"kind": "volume_pullback",
"title": "KOBJ002 synthetic K-line volume pullback support",
"result_file": "kline_volume_pullback_result.csv",
"chart_file": "KOBJ002_volume_pullback.svg",
"closes": [
20.00,
20.25,
20.55,
20.80,
21.00,
21.25,
21.50,
21.72,
21.95,
22.10,
21.95,
21.82,
21.75,
21.90,
22.15,
22.40,
22.62,
22.84,
23.00,
22.88,
23.12,
23.36,
23.30,
23.55,
23.72,
],
"volumes": [
3600,
3720,
3890,
4020,
4160,
4310,
4460,
4590,
4700,
4820,
4380,
3960,
3580,
3440,
3710,
3980,
4200,
4410,
4620,
4210,
4430,
4650,
4240,
4480,
4700,
],
},
]
summaries = [_run_one(config) for config in configs]
print(json.dumps(summaries, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()