50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
"""Phase 3c 模拟盘端到端冒烟(部署后跑)。
|
||
|
||
流程:login → POST /paper/create → GET /paper/{id} → /equity → /trades → /strategies
|
||
用法:python scripts/smoke_phase3c.py https://vnpy.mysanguo.top admin <password>
|
||
"""
|
||
import json
|
||
import sys
|
||
import urllib.request
|
||
|
||
|
||
def _req(method: str, base: str, path: str, token: str | None, body=None):
|
||
url = f"{base}/api/v1{path}"
|
||
headers = {"Content-Type": "application/json"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
data = json.dumps(body).encode() if body is not None else None
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
return json.loads(resp.read())
|
||
|
||
|
||
def main() -> int:
|
||
base = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8000"
|
||
user = sys.argv[2] if len(sys.argv) > 2 else "admin"
|
||
pwd = sys.argv[3] if len(sys.argv) > 3 else "admin"
|
||
|
||
tok = _req("POST", base, "/auth/login", None, {"username": user, "password": pwd})["token"]
|
||
print("✓ login")
|
||
|
||
aid = _req("POST", base, "/paper/create", tok, {
|
||
"mode": "replay", "interval": "d", "symbols": ["600000"],
|
||
"strategies": [{"name": "DoubleMa", "symbol": "600000",
|
||
"match_session": "next_open"}],
|
||
"start": "2024-01-01", "end": "2024-06-30", "initial_capital": 1_000_000,
|
||
})["account_id"]
|
||
print(f"✓ create paper #{aid}")
|
||
|
||
assert _req("GET", base, f"/paper/{aid}", tok)["id"] == aid
|
||
print("✓ get paper")
|
||
assert isinstance(_req("GET", base, f"/paper/{aid}/equity", tok), list)
|
||
assert isinstance(_req("GET", base, f"/paper/{aid}/trades", tok), list)
|
||
assert isinstance(_req("GET", base, f"/paper/{aid}/strategies", tok), list)
|
||
print("✓ equity / trades / strategies")
|
||
print(f"\n✅ Phase 3c 冒烟通过(account #{aid})")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|