yuxuanhui
101 lines
4.6 KiB
Python
101 lines
4.6 KiB
Python
"""Synthetic simulation HTTP used by isolated API and browser acceptance."""
|
|
|
|
import json
|
|
|
|
import httpx
|
|
|
|
|
|
class Platform:
|
|
def __init__(self):
|
|
self.posts = []
|
|
self.existing_alpha_ids = None
|
|
self.simulations = {}
|
|
self.alphas = {}
|
|
self.reject = None
|
|
self.pending = False
|
|
self.detail_fail = False
|
|
self.fail_child = None
|
|
self.missing = False
|
|
self.selection_reads = []
|
|
self.selection_result = {"count": 2, "results": [{"id": "component1", "value": 0.3}, {"id": "component2", "value": 0.7}]}
|
|
self.actual_components = {"count": 2, "results": [{"id": "component1"}, {"id": "component2"}]}
|
|
self.secret = "synthetic-platform-secret"
|
|
|
|
def __call__(self, request):
|
|
path = request.url.path
|
|
if path == "/simulations/super-selection":
|
|
self.selection_reads.append(dict(request.url.params))
|
|
return httpx.Response(200, json=self.selection_result)
|
|
if path == "/authentication":
|
|
return httpx.Response(201, json={})
|
|
if path == "/simulations" and request.method == "POST":
|
|
data = json.loads(request.content)
|
|
data = data if isinstance(data, list) else [data]
|
|
self.posts.append(data)
|
|
if self.reject == "unknown":
|
|
raise httpx.ReadTimeout("synthetic timeout", request=request)
|
|
if self.reject == "session":
|
|
self.reject = None
|
|
return httpx.Response(401)
|
|
if self.reject == "rate":
|
|
return httpx.Response(429, headers={"Retry-After": "0.01"})
|
|
if self.reject == "bad":
|
|
return httpx.Response(400, json={"error": self.secret})
|
|
parent = f"p{len(self.posts)}"
|
|
ids = []
|
|
for i, item in enumerate(data):
|
|
child = parent if len(data) == 1 else f"{parent}c{i}"
|
|
aid = self.existing_alpha_ids[i] if self.existing_alpha_ids and item["type"] != "SUPER" else f"alpha{parent}{i}"
|
|
progress = {
|
|
"status": "COMPLETE",
|
|
"alpha": aid,
|
|
"regular": item.get("regular", ""),
|
|
"settings": item["settings"],
|
|
}
|
|
if i == self.fail_child:
|
|
progress = {
|
|
"status": "FAILED",
|
|
"regular": item.get("regular", ""),
|
|
"settings": item["settings"],
|
|
"message": "invalid expression",
|
|
}
|
|
self.simulations[child] = progress
|
|
self.alphas[aid] = {
|
|
"id": aid,
|
|
"regular": {"code": item.get("regular", "")},
|
|
"type": "REGULAR",
|
|
"settings": item["settings"],
|
|
"is": {"sharpe": None, "fitness": 0.8},
|
|
"status": "UNSUBMITTED",
|
|
}
|
|
if item["type"] == "SUPER":
|
|
assert len(data) == 1, "SUPER must be submitted singly"
|
|
self.simulations[child].update(type="SUPER", selection=item["selection"], combo=item["combo"])
|
|
self.simulations[child].pop("regular", None)
|
|
self.alphas[aid].update(type="SUPER", selection={"code": item["selection"], "description": "Selection rationale"}, combo={"code": item["combo"], "description": "Combo rationale"}, components=self.actual_components)
|
|
self.alphas[aid].pop("regular", None)
|
|
ids.append(child)
|
|
if len(data) > 1:
|
|
self.simulations[parent] = {
|
|
"status": "COMPLETE",
|
|
"children": list(reversed(ids[1:] if self.missing else ids)),
|
|
}
|
|
if self.reject == "missing_location":
|
|
return httpx.Response(201)
|
|
return httpx.Response(
|
|
201, headers={"Location": f"https://api.worldquantbrain.com/simulations/{parent}"}
|
|
)
|
|
if path.startswith("/simulations/"):
|
|
return httpx.Response(
|
|
200, json={"status": "PENDING"} if self.pending else self.simulations[path.rsplit("/", 1)[-1]]
|
|
)
|
|
if path.startswith("/alphas/"):
|
|
if self.detail_fail:
|
|
return httpx.Response(404)
|
|
return httpx.Response(200, json=self.alphas[path.rsplit("/", 1)[-1]])
|
|
if path == "/users/self":
|
|
return httpx.Response(200, json={"id": "TEST_USER"})
|
|
if path.startswith("/users/self/"):
|
|
return httpx.Response(200, json={"results": [], "count": 0})
|
|
raise AssertionError(f"Unexpected HTTP {request.method} {path}")
|