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'<line x1="{cx:.1f}" y1="{y(high):.1f}" x2="{cx:.1f}" y2="{y(low):.1f}" '
|
f'stroke="{color}" stroke-width="1.5" />'
|
)
|
candle_parts.append(
|
f'<rect x="{cx - 5:.1f}" y="{body_y:.1f}" width="10" height="{body_h:.1f}" '
|
f'fill="{color}" opacity="0.78" />'
|
)
|
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'<circle cx="{cx:.1f}" cy="{y(low) - 13:.1f}" r="6" fill="#d92323" />'
|
)
|
signal_parts.append(
|
f'<text x="{cx + 8:.1f}" y="{y(low) - 18:.1f}" font-size="12" fill="#d92323">signal</text>'
|
)
|
|
ma_polyline = ""
|
if len(ma_points) >= 2:
|
ma_polyline = (
|
f'<polyline points="{" ".join(ma_points)}" fill="none" '
|
'stroke="#d49a00" stroke-width="2" />'
|
)
|
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="#fbfaf6" />
|
<text x="{margin}" y="28" font-size="18" font-family="Arial" fill="#222">{title}</text>
|
<line x1="{margin}" y1="{height - margin}" x2="{width - margin}" y2="{height - margin}" stroke="#333" />
|
<line x1="{margin}" y1="{margin}" x2="{margin}" y2="{height - margin}" stroke="#333" />
|
{''.join(candle_parts)}
|
{ma_polyline}
|
{''.join(signal_parts)}
|
<text x="{margin}" y="{height - 14}" font-size="12" font-family="Arial" fill="#555">synthetic K-line data; red dot marks detected signal</text>
|
</svg>
|
"""
|
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()
|
