yuxuanhui
272 lines
12 KiB
Python
272 lines
12 KiB
Python
"""Bounded metadata reads with all-or-nothing publication and separate annotations."""
|
|
|
|
import hashlib
|
|
import json
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from ..models import CatalogResource, OperatorNote, now
|
|
from ..research.serialization import encode_snapshot as jsonable_encoder
|
|
from ..worldquant import WqError
|
|
|
|
|
|
async def upstream(operation):
|
|
try:
|
|
return await operation
|
|
except WqError as exc:
|
|
raise HTTPException(
|
|
409 if exc.code in ("disconnected", "verification_required") else 502, str(exc)
|
|
) from None
|
|
|
|
|
|
def availability_key(field_id, scope):
|
|
return "field:" + hashlib.sha256(f"{field_id}|{scope.key()}".encode()).hexdigest()
|
|
|
|
|
|
def setting_rows(data):
|
|
"""Decode hierarchical OPTIONS choices; incomplete options are not invented."""
|
|
try:
|
|
children = data["actions"]["POST"]["settings"]["children"]
|
|
|
|
def choices(key, instrument=None, region=None, delay=None, universe=None):
|
|
value = children[key]["choices"]
|
|
if isinstance(value, dict) and "instrumentType" in value:
|
|
value = value["instrumentType"][instrument]
|
|
elif isinstance(value, dict) and instrument in value:
|
|
value = value[instrument]
|
|
if isinstance(value, dict) and "region" in value:
|
|
value = value["region"][region]
|
|
for dimension, selected in (("delay", delay), ("universe", universe)):
|
|
if isinstance(value, dict) and dimension in value:
|
|
branches = value[dimension]
|
|
if not isinstance(branches, dict):
|
|
raise ValueError()
|
|
value = branches.get(str(selected), branches.get(selected))
|
|
if not isinstance(value, list):
|
|
raise ValueError()
|
|
return [item["value"] for item in value]
|
|
|
|
rows = []
|
|
for instrument in choices("instrumentType"):
|
|
if instrument != "EQUITY":
|
|
continue
|
|
for region in choices("region", instrument):
|
|
for delay in choices("delay", instrument, region):
|
|
if type(delay) is not int or delay not in (0, 1):
|
|
continue
|
|
for universe in choices("universe", instrument, region, delay):
|
|
neutralizations = (
|
|
choices("neutralization", instrument, region, delay, universe)
|
|
if "neutralization" in children
|
|
else []
|
|
)
|
|
fields = {}
|
|
for key in (
|
|
"decay", "truncation", "pasteurization", "unitHandling",
|
|
"nanHandling", "language", "visualization", "maxTrade", "maxPosition",
|
|
):
|
|
definition = children.get(key)
|
|
if not isinstance(definition, dict):
|
|
continue
|
|
field = {}
|
|
if "choices" in definition:
|
|
field["choices"] = choices(key, instrument, region, delay, universe)
|
|
if any(type(v) not in (str, int, float, bool) for v in field["choices"]):
|
|
raise ValueError()
|
|
for source, target in (("min_value", "minimum"), ("max_value", "maximum")):
|
|
if type(definition.get(source)) in (int, float):
|
|
field[target] = definition[source]
|
|
fields[key] = field
|
|
rows.append(
|
|
{
|
|
"instrument_type": instrument,
|
|
"region": region,
|
|
"universe": universe,
|
|
"delay": delay,
|
|
"neutralizations": neutralizations,
|
|
"fields": fields,
|
|
}
|
|
)
|
|
if not rows:
|
|
raise ValueError()
|
|
return rows
|
|
except (KeyError, TypeError, ValueError):
|
|
raise HTTPException(502, "平台设置结构无法识别，未发布新快照") from None
|
|
|
|
|
|
def normalize_availability(data, *, instrument_type=None):
|
|
"""Use the request's instrument only for the platform field-detail `data` form.
|
|
|
|
Legacy availability rows must still state their own instrument. Missing market,
|
|
delay or universe never inherits the requested scope.
|
|
"""
|
|
raw = data.get("availability")
|
|
detail_form = raw is None and isinstance(data.get("data"), list)
|
|
if detail_form:
|
|
raw = data["data"]
|
|
if not isinstance(raw, list):
|
|
return {"status": "needs_review", "items": [], "reason": "平台未提供可识别的 availability 列表"}
|
|
rows, malformed = [], False
|
|
for item in raw:
|
|
if not isinstance(item, dict):
|
|
malformed = True
|
|
continue
|
|
universes = item.get("universe", item.get("universes"))
|
|
universes = universes if isinstance(universes, list) else [universes]
|
|
for universe in universes:
|
|
if (
|
|
item.get("instrumentType", instrument_type if detail_form else None) == "EQUITY"
|
|
and type(item.get("delay")) is int
|
|
and item["delay"] in (0, 1)
|
|
and isinstance(item.get("region"), str)
|
|
and isinstance(universe, str)
|
|
):
|
|
rows.append(
|
|
{
|
|
"instrument_type": "EQUITY",
|
|
"region": item["region"],
|
|
"delay": item["delay"],
|
|
"universe": universe,
|
|
}
|
|
)
|
|
else:
|
|
malformed = True
|
|
return {
|
|
"status": "available" if rows and not malformed else "needs_review",
|
|
"items": rows,
|
|
"reason": "可用性列表包含不完整项" if malformed else "",
|
|
}
|
|
|
|
|
|
class ResearchMetadata:
|
|
def __init__(self, db, client=None):
|
|
self.db, self.client = db, client
|
|
|
|
async def publish(self, key, kind, content):
|
|
row = await self.db.get(CatalogResource, key)
|
|
if row:
|
|
row.content, row.fetched_at = content, now()
|
|
else:
|
|
row = CatalogResource(key=key, kind=kind, content=content)
|
|
self.db.add(row)
|
|
await self.db.flush()
|
|
return self.output(row)
|
|
|
|
@staticmethod
|
|
def output(row):
|
|
return jsonable_encoder({"key": row.key, "content": row.content, "fetched_at": row.fetched_at})
|
|
|
|
async def get(self, key):
|
|
row = await self.db.get(CatalogResource, key)
|
|
return self.output(row) if row else {"key": key, "content": {}, "fetched_at": None}
|
|
|
|
async def refresh_operators(self):
|
|
items, seen = [], set()
|
|
for offset in range(0, 10000, 100):
|
|
page = await upstream(self.client.operators(offset))
|
|
values = page if isinstance(page, list) else page.get("results")
|
|
if not isinstance(values, list):
|
|
raise HTTPException(502, "算子目录格式无法识别，保留原快照")
|
|
for item in values:
|
|
if not isinstance(item, dict) or not isinstance(item.get("name"), str):
|
|
raise HTTPException(502, "算子目录缺少名称，保留原快照")
|
|
if item["name"] in seen:
|
|
raise HTTPException(502, "算子分页重复，未发布不完整目录")
|
|
seen.add(item["name"])
|
|
items.append(
|
|
{
|
|
key: item.get(key)
|
|
for key in (
|
|
"name",
|
|
"category",
|
|
"description",
|
|
"definition",
|
|
"example",
|
|
"scope",
|
|
"type",
|
|
"parameters",
|
|
)
|
|
}
|
|
)
|
|
if (
|
|
isinstance(page, list)
|
|
or (isinstance(page.get("count"), int) and offset + len(values) >= page["count"])
|
|
or (not page.get("next") and len(values) < 100)
|
|
):
|
|
return await self.publish("operators", "operators", {"items": items})
|
|
if not values:
|
|
raise HTTPException(502, "算子分页提前结束")
|
|
raise HTTPException(502, "算子分页超过本地限制，未发布新快照")
|
|
|
|
async def operators(self, q="", category=None, favorite=False, limit=25, offset=0):
|
|
snapshot = await self.get("operators")
|
|
notes = {r.name: r for r in await self.db.scalars(select(OperatorNote))}
|
|
rows = []
|
|
for item in snapshot["content"].get("items", []):
|
|
note = notes.get(item["name"])
|
|
if q.lower() not in json.dumps(item, ensure_ascii=False).lower() or (
|
|
category and item["category"] != category
|
|
):
|
|
continue
|
|
if favorite and not (note and note.favorite):
|
|
continue
|
|
rows.append(
|
|
{
|
|
**item,
|
|
"local": {
|
|
"note": note.note if note else "",
|
|
"favorite": note.favorite if note else False,
|
|
"version": note.version if note else 0,
|
|
},
|
|
}
|
|
)
|
|
return {
|
|
"items": rows[offset : offset + limit],
|
|
"total": len(rows),
|
|
"limit": limit,
|
|
"offset": offset,
|
|
"fetched_at": snapshot["fetched_at"],
|
|
"categories": sorted(
|
|
{str(i.get("category")) for i in snapshot["content"].get("items", []) if i.get("category")}
|
|
),
|
|
}
|
|
|
|
async def annotate(self, name, body):
|
|
snapshot = await self.get("operators")
|
|
if name not in {i["name"] for i in snapshot["content"].get("items", [])}:
|
|
raise HTTPException(404, "算子不在已同步目录中")
|
|
if body.version == 0:
|
|
if await self.db.get(OperatorNote, name):
|
|
raise HTTPException(409, "备注已变化")
|
|
self.db.add(OperatorNote(name=name, note=body.note, favorite=body.favorite))
|
|
try:
|
|
await self.db.flush()
|
|
except IntegrityError:
|
|
raise HTTPException(409, "备注已变化，请保留草稿并刷新") from None
|
|
else:
|
|
result = await self.db.execute(
|
|
update(OperatorNote)
|
|
.where(OperatorNote.name == name, OperatorNote.version == body.version)
|
|
.values(note=body.note, favorite=body.favorite, version=body.version + 1)
|
|
)
|
|
if result.rowcount != 1:
|
|
raise HTTPException(409, "备注已变化，请保留草稿并刷新")
|
|
return {"ok": True, "version": body.version + 1}
|
|
|
|
async def refresh_settings(self):
|
|
data = await upstream(self.client.research_setting_options())
|
|
return await self.publish("settings", "settings", {"items": setting_rows(data)})
|
|
|
|
async def refresh_availability(self, body):
|
|
data = await upstream(self.client.field_availability(body.field_id, body.scope))
|
|
if data.get("id") is not None and data["id"] != body.field_id:
|
|
raise HTTPException(502, "平台返回字段与请求不一致，保留原可用性快照")
|
|
content = {
|
|
**normalize_availability(data, instrument_type=body.scope.instrument_type),
|
|
"field_id": body.field_id,
|
|
"scope": body.scope.model_dump(),
|
|
}
|
|
return await self.publish(availability_key(body.field_id, body.scope), "availability", content)
|