"""Offline academy lab: synthetic data ONLY. No network, credentials or orders.
Python 3.10+. python research_lab.py --output results
Execution convention: prior closed-bar SMA -> next open, fixed units, no leverage.
Costs are a synthetic one-way bps charge, NOT a real broker FX contract model.
"""
import argparse
import csv
import datetime as dt
import hashlib
import html
import json
import math
from pathlib import Path
import random
import statistics
ROOT = Path(__file__).resolve().parent
def read_prices(path):
with Path(path).open(encoding="utf-8-sig", newline="") as stream:
reader = csv.DictReader(stream)
required = {"timestamp", "open", "high", "low", "close", "volume"}
if not required.issubset(reader.fieldnames or []):
raise ValueError("Missing columns: " + ", ".join(sorted(required - set(reader.fieldnames or []))))
rows, seen, rejected, duplicates = [], set(), [], 0
for line, row in enumerate(reader, 2):
try:
timestamp = dt.datetime.fromisoformat(row["timestamp"].replace("Z", "+00:00"))
if timestamp.tzinfo is None:
raise ValueError("timestamp needs timezone")
timestamp = timestamp.astimezone(dt.timezone.utc).isoformat()
values = {k: float(row[k]) for k in required - {"timestamp"}}
if not all(math.isfinite(v) for v in values.values()):
raise ValueError("non-finite number")
o, h, l, c = (values[k] for k in ("open", "high", "low", "close"))
if min(o, h, l, c) <= 0 or h < max(o, c) or l > min(o, c) or h < l or values["volume"] < 0:
raise ValueError("invalid OHLC/volume")
if timestamp in seen:
duplicates += 1
continue
seen.add(timestamp)
rows.append({"timestamp": timestamp, **values})
except (ValueError, TypeError) as exc:
rejected.append({"line": line, "reason": str(exc)})
rows.sort(key=lambda r: r["timestamp"])
if not rows:
raise ValueError("No valid price rows")
return rows, {"valid": len(rows), "duplicates": duplicates, "rejected": rejected}
def max_drawdown(equity):
if not equity or any(not math.isfinite(x) or x <= 0 for x in equity):
raise ValueError("Equity must be positive and finite")
peak, worst = equity[0], 0.0
for value in equity:
peak = max(peak, value)
worst = max(worst, 1 - value / peak)
return worst
def backtest(rows, fast=3, slow=8, cost_bps=1.0):
if not 0 < fast < slow or not math.isfinite(cost_bps) or cost_bps < 0:
raise ValueError("Require 0 < fast < slow and nonnegative finite costs")
cash, units, costs = 10000.0, 0, 0.0
closes, curve, orders = [], [], []
for index, row in enumerate(rows):
# At open of this bar, ONLY closes before this bar are available.
target = 0
if len(closes) >= slow:
short = statistics.mean(closes[-fast:])
long = statistics.mean(closes[-slow:])
target = 1000 if short > long else 0
change = target - units
if change:
cost = abs(change) * row["open"] * cost_bps / 10000
if change > 0 and cash < change * row["open"] + cost:
raise ValueError("Insufficient cash; leverage is not supported")
cash -= change * row["open"] + cost
costs += cost
orders.append({"time": row["timestamp"], "signal_time": rows[index - 1]["timestamp"],
"units_change": change, "price": row["open"], "cost": cost})
units = target
curve.append({"timestamp": row["timestamp"], "equity": cash + units * row["close"], "units": units})
closes.append(row["close"])
# Include starting cash in the drawdown calculation; no artificial forced liquidation.
return {"initial_equity": 10000.0, "final_equity": curve[-1]["equity"],
"return": curve[-1]["equity"] / 10000 - 1, "costs": costs,
"max_drawdown": max_drawdown([10000] + [r["equity"] for r in curve]),
"open_units": units, "open_units_valuation": "last close; not realized cash",
"order_events": len(orders), "curve": curve, "orders": orders}
def tail_risk(losses, confidence=0.95):
if not losses or not 0 < confidence < 1 or any(not math.isfinite(x) for x in losses):
raise ValueError("Finite losses and 0 < confidence < 1 required")
ordered = sorted(losses)
rank = math.ceil(confidence * len(ordered)) - 1
count = max(1, math.ceil(round((1 - confidence) * len(ordered), 10)))
return {"var_nearest_rank": ordered[rank], "es_worst_tail_mean": statistics.mean(ordered[-count:]),
"tail_count": count, "n": len(ordered)}
def pretrade(existing, additional, limit, age_seconds, max_age=30):
values = [existing, additional, limit, age_seconds, max_age]
if not all(math.isfinite(x) for x in values) or min(values) < 0 or limit == 0:
return "reject_invalid_input"
if age_seconds > max_age:
return "reject_stale_quote"
if existing + additional > limit:
return "reject_exposure"
return "accept_simulation_only"
def trade_report(path):
seen, accepted, rejected, duplicates = set(), [], [], 0
with Path(path).open(encoding="utf-8-sig", newline="") as stream:
for line, row in enumerate(csv.DictReader(stream), 2):
try:
identity = row["deal_id"].strip()
if not identity:
raise ValueError("missing deal ID")
timestamp = dt.datetime.fromisoformat(row["timestamp"].replace("Z", "+00:00"))
if timestamp.tzinfo is None:
raise ValueError("timezone missing")
values = [float(row[k]) for k in ("gross", "commission", "swap")]
if not all(math.isfinite(x) for x in values):
raise ValueError("non-finite P&L")
if identity in seen:
duplicates += 1
continue
seen.add(identity)
accepted.append({"deal_id": identity, "timestamp": timestamp.isoformat(), "net": sum(values)})
except (ValueError, TypeError, KeyError) as exc:
rejected.append({"line": line, "reason": str(exc)})
accepted.sort(key=lambda r: (r["timestamp"], r["deal_id"]))
profits = [r["net"] for r in accepted]
wins, losses = sum(x for x in profits if x > 0), -sum(x for x in profits if x < 0)
equity = [1000.0]
for value in profits:
equity.append(equity[-1] + value)
return {"valid": len(accepted), "duplicates": duplicates, "rejected": rejected,
"net_profit": sum(profits), "profit_factor": wins / losses if losses else None,
"win_rate_including_flat": sum(x > 0 for x in profits) / len(profits) if profits else None,
"max_drawdown": max_drawdown(equity), "rows": accepted}
def self_test():
assert abs(max_drawdown([10000, 11000, 9900, 10100]) - 0.1) < 1e-12
assert tail_risk([0] * 18 + [2, 8])["var_nearest_rank"] == 2
assert tail_risk([0] * 18 + [2, 8])["es_worst_tail_mean"] == 8
assert pretrade(600, 500, 1000, 1) == "reject_exposure"
assert pretrade(0, 1, 1000, 301) == "reject_stale_quote"
assert pretrade(0, float("nan"), 1000, 1) == "reject_invalid_input"
assert pretrade(600, 100, 1000, 1) == "accept_simulation_only"
rows, _ = read_prices(ROOT / "ohlcv-synthetic.csv")
one = backtest(rows)
# Causality: changing future bars must not alter earlier equity or orders.
prefix = backtest(rows[:100])
assert prefix["curve"] == one["curve"][:100]
assert all(o["signal_time"] < o["time"] for o in one["orders"])
assert backtest(rows, cost_bps=3)["final_equity"] <= backtest(rows, cost_bps=0)["final_equity"]
assert one == backtest(rows)
print("PASS: arithmetic, tail convention, risk rejection, causality, determinism, cost sensitivity")
def run(output):
rows, quality = read_prices(ROOT / "ohlcv-dirty.csv")
result = backtest(rows)
dest = Path(output)
dest.mkdir(parents=True, exist_ok=True)
with (dest / "equity.csv").open("w", encoding="utf-8", newline="") as stream:
writer = csv.DictWriter(stream, fieldnames=["timestamp", "equity", "units"])
writer.writeheader(); writer.writerows(result["curve"])
artifact = {"data_kind": "synthetic teaching data; not live or historical market observations",
"execution": "prior closed-bar SMA -> next open; fixed 1000 units; cash only",
"input_sha256": hashlib.sha256((ROOT / "ohlcv-dirty.csv").read_bytes()).hexdigest(),
"parameters": {"fast": 3, "slow": 8, "one_way_cost_bps": 1},
"quality": quality, "research": {k: v for k, v in result.items() if k != "curve"},
"trades": trade_report(ROOT / "trades-dirty.csv"),
"tail_example": tail_risk([0] * 18 + [2, 8]),
"grid_example": {"inventory": 3, "average_entry": 99, "mark": 95,
"unrealized": sum(95 - x for x in [100, 99, 98]), "realized": 2},
"limitations": ["No FX margin, financing, order book, stop-order or live API model",
"No investment recommendation, performance claim or production deployment"]}
text = json.dumps(artifact, ensure_ascii=False, indent=2, allow_nan=False)
(dest / "report.json").write_text(text, encoding="utf-8")
values = [r["equity"] for r in result["curve"]]
low, high = min(values), max(values)
points = " ".join(f"{i * 800 / (len(values)-1):.2f},{190-(v-low)/max(high-low,1e-12)*170:.2f}" for i,v in enumerate(values))
(dest / "report.html").write_text('离线教学报告