wq_test_script.py

import json, time, requests, numpy as np
from requests.auth import HTTPBasicAuth

with open("credential.txt") as f:
    raw = f.read().strip()
    if raw.startswith("1|"): raw = raw[2:]
    cred = json.loads(raw)

s = requests.Session()
s.auth = HTTPBasicAuth(cred[0], cred[1])
s.headers.update({"Content-Type":"application/json","Accept":"application/json"})
r = s.post("https://api.worldquantbrain.com/authentication")
print(f"AUTH: {r.status_code}")

# List alphas
all_a, off = [], 0
while True:
    data = s.get(f"https://api.worldquantbrain.com/users/self/alphas", params={"limit":100,"offset":off}).json()
    batch = data.get("results", data.get("alphas", []))
    if not batch: break
    all_a.extend(batch)
    if len(batch) < 100: break
    off += 100
    time.sleep(1)
print(f"Total: {len(all_a)}")
active = [a for a in all_a if a.get("status")=="ACTIVE"]
print(f"Active: {len(active)}")
for a in active:
    is_ = a.get("is",{})
    print(f"  {a.get('id')}: S={is_.get('sharpe',0):.3f} F={is_.get('fitness',0):.3f} TO={is_.get('turnover',1):.3f}")

# Get PnL for active (just first 3)
def get_pnl(s, aid):
    r = s.get(f"https://api.worldquantbrain.com/alphas/{aid}/recordsets/pnl")
    if r.status_code != 200: return []
    data = r.json()
    props = data.get("schema",{}).get("properties",[])
    if isinstance(props, list): idx = next((i for i,p in enumerate(props) if p.get("name","").lower() in ("pnl","cum_pnl")), 1)
    else: idx = next((v["index"] for k,v in props.items() if k.lower() in ("pnl","cum_pnl")), 1)
    out = []
    for row in data.get("records",[]):
        rec = row[0] if isinstance(row,list) and len(row)==1 and isinstance(row[0],list) else row
        try: out.append(float(rec[idx]))
        except: continue
    return out

pnl_map = {}
for a in active[:5]:
    p = get_pnl(s, a["id"])
    if len(p) > 20: pnl_map[a["id"]] = p
    time.sleep(0.5)
print(f"PnL fetched for {len(pnl_map)} active alphas")

# Run 1 quick simulation first
sim = ("FCF Yield", "group_rank(ts_rank(free_cash_flow_reported_value / equity, 126), subindustry)", {"decay":0,"neutralization":"SUBINDUSTRY"})
name, expr, extra = sim
base = {"instrumentType":"EQUITY","region":"USA","universe":"TOP3000","delay":1,"truncation":0.08,"pasteurization":"ON","unitHandling":"VERIFY","nanHandling":"ON","language":"FASTEXPR","visualization":False}
base.update(extra)
payload = {"type":"REGULAR","settings":base,"regular":expr}
print(f"\nRunning: {name}")
r2 = s.post("https://api.worldquantbrain.com/simulations", json=payload)
print(f"POST: {r2.status_code}")
if r2.status_code == 201:
    sim_id = r2.headers["Location"].rstrip("/").split("/")[-1]
    print(f"sim_id={sim_id}")
    for p in range(40):
        time.sleep(8)
        d = s.get(f"https://api.worldquantbrain.com/simulations/{sim_id}").json()
        st = d.get("status")
        if st == "COMPLETE":
            aid = d["alpha"]
            a = s.get(f"https://api.worldquantbrain.com/alphas/{aid}").json()
            is_ = a.get("is",{})
            print(f"RESULT: S={is_.get('sharpe',0):.3f} F={is_.get('fitness',0):.3f} TO={is_.get('turnover',1):.3f} DD={is_.get('drawdown',0):.3f}")
            for c in is_.get("checks",[]):
                print(f"  CHECK: {c['name']}={c.get('result')}")
            break
        elif st in ("ERROR","FAILED"):
            print(f"ERROR: {st}")
            break
        if p%3==0: print(f"poll[{p}]: {st}")
else:
    print(f"FAIL: {r2.text[:300]}")
print("DONE")