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]
|
EXPERIMENT_ID = "EXP-20260601-TIMESERIES-SMOKE-001"
|
RUN_ID = "RUN-20260601-TIMESERIES-SMOKE-001"
|
OBJECT_ID = "OBJECT001"
|
|
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
|
|
|
def rel(path: Path) -> str:
|
return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
|
|
|
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 business_days(start: date, n: int) -> list[date]:
|
days: list[date] = []
|
current = start
|
while len(days) < n:
|
if current.weekday() < 5:
|
days.append(current)
|
current += timedelta(days=1)
|
return days
|
|
|
def build_synthetic_ohlc() -> list[dict[str, object]]:
|
days = business_days(date(2026, 5, 1), 20)
|
closes = [
|
10.00,
|
9.82,
|
9.75,
|
9.68,
|
9.70,
|
9.77,
|
9.88,
|
10.05,
|
10.28,
|
10.62,
|
10.55,
|
10.74,
|
10.90,
|
10.86,
|
11.05,
|
11.22,
|
11.18,
|
11.36,
|
11.50,
|
11.42,
|
]
|
rows: list[dict[str, object]] = []
|
prev_close = closes[0]
|
for idx, (event_date, close) in enumerate(zip(days, closes), start=1):
|
open_price = prev_close * (1 + (0.002 if idx % 2 == 0 else -0.001))
|
high = max(open_price, close) * 1.018
|
low = min(open_price, close) * 0.985
|
volume = 100000 + idx * 6500 + (45000 if idx in {9, 10, 12} else 0)
|
rows.append(
|
{
|
"experiment_id": EXPERIMENT_ID,
|
"run_id": RUN_ID,
|
"object_id": OBJECT_ID,
|
"event_date": event_date.isoformat(),
|
"open": round(open_price, 2),
|
"high": round(high, 2),
|
"low": round(low, 2),
|
"close": round(close, 2),
|
"volume": int(volume),
|
"data_source": "SYNTHETIC_TIMESERIES_SMOKE",
|
"source_visible_at": f"{event_date.isoformat()} 15:00:00",
|
}
|
)
|
prev_close = close
|
return rows
|
|
|
def add_ma_and_signal(rows: list[dict[str, object]]) -> list[dict[str, object]]:
|
closes = [float(row["close"]) for row in rows]
|
output: list[dict[str, object]] = []
|
prev_close = None
|
prev_ma5 = None
|
for idx, row in enumerate(rows):
|
if idx >= 4:
|
ma5 = round(sum(closes[idx - 4 : idx + 1]) / 5, 4)
|
else:
|
ma5 = None
|
close = float(row["close"])
|
breakout = bool(ma5 is not None and prev_ma5 is not None and close > ma5 and prev_close is not None and prev_close <= prev_ma5)
|
output.append(
|
{
|
**row,
|
"ma5": "" if ma5 is None else ma5,
|
"ma5_breakout_flag": int(breakout),
|
"signal_rule": "close_cross_above_ma5_after_pullback",
|
}
|
)
|
prev_close = close
|
prev_ma5 = ma5
|
return output
|
|
|
def write_csv(path: Path, rows: list[dict[str, object]]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def write_svg(path: Path, rows: list[dict[str, object]]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
width = 920
|
height = 420
|
margin = 50
|
plot_h = height - margin * 2
|
prices = [float(row[k]) for row in rows for k in ("high", "low")]
|
min_p = min(prices)
|
max_p = max(prices)
|
|
def y(price: float) -> float:
|
return margin + (max_p - price) / (max_p - min_p) * plot_h
|
|
step = (width - margin * 2) / len(rows)
|
parts = [
|
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="50" y="28" font-size="18" font-family="Consolas, monospace">Timeseries smoke experiment: OBJECT001 / MA5 breakout</text>',
|
f'<line x1="{margin}" y1="{height-margin}" x2="{width-margin}" y2="{height-margin}" stroke="#777"/>',
|
f'<line x1="{margin}" y1="{margin}" x2="{margin}" y2="{height-margin}" stroke="#777"/>',
|
]
|
for idx, row in enumerate(rows):
|
x = margin + idx * step + step / 2
|
open_p = float(row["open"])
|
close_p = float(row["close"])
|
high_p = float(row["high"])
|
low_p = float(row["low"])
|
color = "#c0392b" if close_p >= open_p else "#1f6f50"
|
body_top = min(y(open_p), y(close_p))
|
body_h = max(abs(y(open_p) - y(close_p)), 2)
|
parts.append(f'<line x1="{x:.2f}" y1="{y(high_p):.2f}" x2="{x:.2f}" y2="{y(low_p):.2f}" stroke="{color}" stroke-width="2"/>')
|
parts.append(f'<rect x="{x - step * 0.25:.2f}" y="{body_top:.2f}" width="{step * 0.5:.2f}" height="{body_h:.2f}" fill="{color}" opacity="0.85"/>')
|
if int(row["ma5_breakout_flag"]) == 1:
|
parts.append(f'<circle cx="{x:.2f}" cy="{y(close_p) - 12:.2f}" r="6" fill="#2563eb"/>')
|
parts.append(f'<text x="{x + 8:.2f}" y="{y(close_p) - 16:.2f}" font-size="12" fill="#2563eb">breakout</text>')
|
ma_points = []
|
for idx, row in enumerate(rows):
|
if row["ma5"] != "":
|
x = margin + idx * step + step / 2
|
ma_points.append(f'{x:.2f},{y(float(row["ma5"])):.2f}')
|
if ma_points:
|
parts.append(f'<polyline points="{" ".join(ma_points)}" fill="none" stroke="#f59e0b" stroke-width="2"/>')
|
parts.append('<text x="760" y="58" font-size="13" fill="#f59e0b">MA5</text>')
|
parts.append("</svg>")
|
path.write_text("\n".join(parts), encoding="utf-8")
|
|
|
def main() -> None:
|
RAW_DIR.mkdir(parents=True, exist_ok=True)
|
RESULT_DIR.mkdir(parents=True, exist_ok=True)
|
IMG_DIR.mkdir(parents=True, exist_ok=True)
|
|
raw_rows = build_synthetic_ohlc()
|
result_rows = add_ma_and_signal(raw_rows)
|
signal_count = sum(int(row["ma5_breakout_flag"]) for row in result_rows)
|
|
raw_csv = RAW_DIR / "synthetic_timeseries_input.csv"
|
result_csv = RESULT_DIR / "timeseries_signal_result.csv"
|
chart_svg = IMG_DIR / "OBJECT001_2026-05_timeseries_ma5_breakout.svg"
|
summary_json = RESULT_DIR / "summary.json"
|
readout_md = RESULT_DIR / "readout.md"
|
input_manifest = RESULT_DIR / "input_manifest.csv"
|
output_manifest = RESULT_DIR / "output_manifest.csv"
|
|
write_csv(raw_csv, raw_rows)
|
write_csv(result_csv, result_rows)
|
write_svg(chart_svg, result_rows)
|
|
summary = {
|
"experiment_id": EXPERIMENT_ID,
|
"run_id": RUN_ID,
|
"status": "PASS__ENVIRONMENT_SMOKE_READY",
|
"data_policy": "synthetic data; environment smoke only; not a business conclusion",
|
"row_count": len(raw_rows),
|
"object_count": 1,
|
"signal_count": signal_count,
|
"raw_input": rel(raw_csv),
|
"result_table": rel(result_csv),
|
"chart": rel(chart_svg),
|
"future_function_requirement": "not applicable for synthetic environment smoke; no business conclusion",
|
}
|
summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
readout_md.write_text(
|
"\n".join(
|
[
|
"# Timeseries smoke experiment readout",
|
"",
|
f"experiment_id: {EXPERIMENT_ID}",
|
f"run_id: {RUN_ID}",
|
"",
|
"结论：实验环境闭环通过。合成时序数据、结果表、SVG 图、summary 和 manifest 均已生成。",
|
"",
|
"边界：本实验只验证实验体系能记录、执行、归档和审计一个轻量时序图实验，不证明任何业务规则有效。",
|
]
|
),
|
encoding="utf-8",
|
)
|
|
input_rows = [
|
{
|
"artifact": "synthetic_timeseries_input.csv",
|
"path": rel(raw_csv),
|
"artifact_role": "raw_input",
|
"row_count": len(raw_rows),
|
"sha256": sha256_file(raw_csv),
|
}
|
]
|
write_csv(input_manifest, input_rows)
|
|
manifest_rows = []
|
for path, role, rows in [
|
(raw_csv, "raw_input", len(raw_rows)),
|
(result_csv, "result_table", len(result_rows)),
|
(chart_svg, "chart", 1),
|
(summary_json, "summary", 1),
|
(readout_md, "readout", 1),
|
(input_manifest, "input_manifest", len(input_rows)),
|
(Path(__file__), "runner_script", 1),
|
]:
|
manifest_rows.append(
|
{
|
"artifact": path.name,
|
"path": rel(path),
|
"artifact_role": role,
|
"row_count_or_count": rows,
|
"sha256": sha256_file(path),
|
}
|
)
|
write_csv(output_manifest, manifest_rows)
|
|
|
if __name__ == "__main__":
|
main()
|
