brain-login-and-sim 0.1.5

WorldQuant BRAIN client — login (biometric Scan + auto re-login), simulation, batch run, and auto-keeping promising alphas

brain-login-and-sim

A Python client for WorldQuant BRAIN. It logs you in (handling the biometric "Scan" step and automatic re-login), runs simulations, runs them in batches with concurrency and resume, and automatically keeps the alphas that pass your criteria.

---

Table of contents

• Features
• Install
• Quick start
• Configuration (.env)
• 1. Logging in
• 2. Running a single simulation
• 3. Running a batch
• 4. Input formats
• 5. Keeping promising alphas
• 6. Resume (skip already-run alphas)
• 7. Rate-limit handling
• 8. Telegram / email notifications
• Full API reference
• Output files
• Complete example
• Testing
• License

---

Features

• Login + biometric Scan — sends the scan link to Telegram/email and polls until you finish scanning on your phone; no need to watch the terminal.
• Auto re-login — when the session expires (~4 hours), a 401 triggers an automatic re-login and the request is retried.
• Rate-limit handling — proactive pacing (pause every N alphas) and reactive 429 auto-backoff (respects Retry-After).
• Promising-alpha filter — keeps alphas with |Sharpe| >= 1 and |Fitness| >= 0.9 (absolute value, so it also flags "flip candidates" that become good when negated).
• Batch — up to 3 concurrent simulations (the platform limit); one failing alpha never aborts the batch.
• Resume — checkpoints by hash; re-running the same file continues where it left off.
• Live feedback — prints which alpha is running and can ping Telegram on every hit.
• Flexible input — a list, a .txt file, or a .jsonl file with expression (+hash).

---

Install

pip install brain-login-and-sim

Requires Python 3.9+. Dependencies (requests, python-dotenv) install automatically.

---

Quick start

from brain_login_and_sim import BrainClient, Simulator

client = BrainClient(notify_telegram=True).authenticate()
sim = Simulator(client, notify_promising=True)

sim.simulate_batch("alphas.jsonl", save_promising=True)

That single call logs in, runs every alpha in the file (3 at a time), keeps the good ones in promising_alphas.jsonl, can resume if interrupted, and pings Telegram on hits.

---

Configuration (.env)

Create a file named .env next to your script (loaded automatically):

# Required — WorldQuant login
WQ_EMAIL=[e-mail]
WQ_PASSWORD=your-password

# Optional — Telegram (for the Scan link and promising-alpha alerts)
TELEGRAM_BOT_TOKEN=123456789:AAH...
TELEGRAM_CHAT_ID=123456789

# Optional — email instead of Telegram
SMTP_USER=[e-mail]
SMTP_PASSWORD=your-app-password
WQ_NOTIFY_EMAIL=[e-mail]

Variable	Purpose
WQ_EMAIL, WQ_PASSWORD	WorldQuant credentials
WQ_CREDENTIALS_FILE	Optional path to a JSON {"email":..., "password":...}
TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID	Telegram bot + chat to message
SMTP_USER, SMTP_PASSWORD, WQ_NOTIFY_EMAIL	Email sending (Gmail App Password)

Telegram setup: message @BotFather → /newbot → get the token; send your bot any message; open https://api.telegram.org/bot<TOKEN>/getUpdates and read chat.id.

---

1. Logging in

from brain_login_and_sim import BrainClient

client = BrainClient(notify_telegram=True).authenticate()
print(client.whoami()["id"])

BrainClient(...) options:

Param	Default	Meaning
email, password	None	Override .env; if omitted, read from env/JSON
notify_telegram	False	Send the biometric Scan link to Telegram
notify_email	False	Send the Scan link by email
poll_interval	5.0	Seconds between polls while waiting for the Scan
biometric_timeout	600.0	Max seconds to wait for the Scan before failing

Useful attributes you can tweak after construction: client.auto_relogin (default True), client.max_429_retries (5), client.backoff_base (5.0), client.backoff_max (120.0).

Biometric Scan: if WorldQuant asks for a Scan, the link is sent to Telegram/email and the client waits, polling until you finish on your phone — then it continues automatically. If you don't scan within biometric_timeout, it raises BrainAuthError; just re-run later (once you've scanned, the next login goes straight through).

Custom Scan handling:

client.authenticate(on_biometric=lambda url: print("Scan here:", url))

---

2. Running a single simulation

from brain_login_and_sim import Simulator

sim = Simulator(client)
r = sim.simulate("ts_rank(close, 5)")

print(r["simulation"]["status"])   # "COMPLETE"
print(r["alpha_id"])               # e.g. "3qAEjo5P"
print(r["alpha"]["is"]["sharpe"])  # in-sample Sharpe
print(r["promising"])              # passes the criteria?

simulate(...) returns a dict:

Key	Meaning
simulation	Raw result of the simulation (has status, and alpha id when complete)
alpha_id	The resulting alpha id (or None if it failed)
alpha	Full alpha detail incl. is (in-sample) / os (out-of-sample) metrics
promising	True if it meets min_sharpe / min_fitness
saved	True if it was written to the promising file (when save_promising=True)

Key simulate params: settings (dict, defaults to USA/TOP3000/delay-1), save_promising (default False), min_sharpe (1.0), min_fitness (0.9), include_inverse (True), timeout (600.0).

---

3. Running a batch

sim = Simulator(client, notify_promising=True)

results = sim.simulate_batch(
    "alphas.jsonl",
    save_promising=True,
    max_concurrent=3,        # up to 3 at once (the platform limit)
    pause_every=20,          # pause after every 20 alphas...
    pause_seconds=300,       # ...for 5 minutes (rate-limit pacing)
)

simulate_batch(expressions, ...) runs every alpha and returns a list of records, one per alpha, in input order:

Field	Meaning
regular	The alpha expression
source_hash	The hash from the JSONL line (if any)
ok	True if it completed successfully
status	COMPLETE / ERROR / FAIL / ...
alpha_id, sharpe, fitness	Result + metrics (if completed)
promising	Met the criteria
saved	Newly written to the promising file
skipped	Skipped because already processed (resume)
error	Error message (e.g. a syntax error in the expression)

All simulate_batch params:

Param	Default	Meaning
expressions	—	File path, list of strings, or list of {expression, hash} dicts
save_promising	True	Keep alphas that pass the criteria
save_path	"promising_alphas.jsonl"	Where to store kept alphas
min_sharpe / min_fitness	1.0 / 0.9	Thresholds
include_inverse	True	Use absolute value (catch flip candidates)
max_concurrent	3	Concurrency, capped at 3
stop_on_error	False	Stop on first error (only in max_concurrent=1 mode)
skip_processed	True	Resume — skip alphas already run
processed_path	"processed.jsonl"	Resume checkpoint file
pause_every	0	Pause after this many alphas (0 = never)
pause_seconds	0.0	How long to pause between chunks
timeout	600.0	Per-simulation timeout
on_progress	None	callback(record) after each alpha (replaces the default print)

A failing alpha (e.g. bad syntax) is recorded with ok=False and its error, and the batch keeps going to the next one.

---

4. Input formats

A list of expressions:

sim.simulate_batch(["rank(close - open)", "ts_rank(close, 5)"])

A .txt file (one expression per line; blank lines and # comments ignored):

rank(close - open)
ts_rank(close, 5)

A .jsonl file (one JSON object per line, must have an expression field):

{"hash": "f7b3...", "expression": "rank(reverse(ts_delta(anl4_cfo_median, 10)))", "status": "generated"}
{"hash": "a1c2...", "expression": "ts_rank(close, 5)", "status": "generated"}

The hash is carried through, so when an alpha passes it's stored as source_hash, letting you trace each kept alpha back to its source line.

You can also load items yourself:

items = Simulator.load_alpha_items("alphas.jsonl")  # [{"expression":..., "source":{...}}, ...]
exprs = Simulator.load_expressions("alphas.txt")    # ["rank(...)", ...]

---

5. Keeping promising alphas

An alpha is "promising" when |Sharpe| >= min_sharpe and |Fitness| >= min_fitness (defaults 1.0 / 0.9). Absolute value means a strongly negative alpha also passes — it becomes good once you negate the expression. Such alphas are flagged inverse=True and get a ready-to-use regular_inverse (e.g. -(rank(close - open))).

# Check / save manually
sim.is_promising(r["alpha"])                       # True / False
sim.save_if_promising(r["alpha"], source=item["source"])

# Read everything kept so far
for a in Simulator.load_promising():
    flip = " [negate first]" if a["inverse"] else ""
    print(a["alpha_id"], a["sharpe"], a["fitness"], a["regular"], flip)

To match WorldQuant's real submission bar, pass stricter thresholds:

sim.simulate_batch("alphas.jsonl", save_promising=True, min_sharpe=1.25, min_fitness=1.0)

To keep only genuinely positive alphas (ignore flip candidates): include_inverse=False.

---

6. Resume (skip already-run alphas)

With skip_processed=True (default), every finished alpha is checkpointed to processed.jsonl (keyed by hash, or by the expression text if there's no hash). Re-running the same file skips what's already done and continues — and because it checkpoints after each alpha, it survives an interruption (Ctrl-C, kill, power loss).

sim.simulate_batch("alphas.jsonl", save_promising=True)   # resumes automatically
sim.simulate_batch("alphas.jsonl", skip_processed=False)  # force a full re-run

To retry specific alphas, delete their lines from processed.jsonl.

---

7. Rate-limit handling

Two layers protect against 429 Too Many Requests:

Proactive pacing — pause between chunks:

sim.simulate_batch("alphas.jsonl", pause_every=20, pause_seconds=300)  # 5 min every 20

Reactive backoff — if a request still returns 429, the client waits (honoring the server's Retry-After, or exponential backoff 5 → 10 → 20 … ≤ 120s) and retries, up to client.max_429_retries times. A rate limit therefore doesn't fail your run. Tune via:

client.max_429_retries = 8
client.backoff_base = 10
client.backoff_max = 300

---

8. Telegram / email notifications

# Scan link on login:
client = BrainClient(notify_telegram=True).authenticate()   # or notify_email=True

# Alert on every promising alpha during sims:
sim = Simulator(client, notify_promising=True)

# Send your own message any time:
from brain_login_and_sim import send_telegram
send_telegram("Batch finished ✅")

Simulator(client, verbose=True) (default) prints now simulating: <expr> for each alpha; set verbose=False to silence it.

---

Full API reference

Module-level

Name	Description
BrainClient	The login/session client
Simulator	Simulation + filtering, takes a logged-in client
BrainAuthError	Raised on auth/Telegram failures
send_telegram(text, bot_token=None, chat_id=None, disable_preview=True)	Send any Telegram message
send_biometric_email(scan_url, ...)	Email the Scan link
send_biometric_telegram(scan_url, ...)	Telegram the Scan link
__version__	Package version string

BrainClient

Method / attr	Description
authenticate(on_biometric=None)	Log in; returns the client
whoami()	Current user info (confirms the session works)
get(path, **kw) / post(path, **kw)	Authenticated request with auto re-login + 429 backoff
is_authenticated	Whether a session cookie is present
auto_relogin, max_429_retries, backoff_base, backoff_max	Resilience knobs

Simulator(client, notify_promising=False, verbose=True)

Method	Description
simulate(regular, ...)	Run one alpha, poll to completion, fetch metrics
simulate_batch(expressions, ...)	Run many (concurrent, resume, pacing, auto-keep)
is_promising(alpha, min_sharpe=1.0, min_fitness=0.9, include_inverse=True)	Test the criteria (static)
save_if_promising(alpha, path=..., source=None)	Append to the promising file if it passes
load_promising(path="promising_alphas.jsonl")	Read all kept alphas (static)
get_operators()	List available operators
get_data_fields(region="USA", delay=1, universe="TOP3000")	List data fields
get_alpha(alpha_id)	Fetch full alpha detail
load_alpha_items(path) / load_expressions(path)	Read alphas from a file

---

Output files

Both are created automatically in your current directory (or wherever save_path / processed_path point — the parent folder must already exist).

• promising_alphas.jsonl — one kept alpha per line: savedAt, source_hash, alpha_id, regular, inverse, regular_inverse, sharpe, fitness, turnover, returns, drawdown, margin, region, universe, delay, neutralization, dateCreated
• processed.jsonl — the resume checkpoint (one line per finished alpha).

---

Complete example

from brain_login_and_sim import BrainClient, Simulator

# 1) Log in (Scan link goes to Telegram; auto re-login if the session expires)
client = BrainClient(notify_telegram=True).authenticate()
print("Logged in as:", client.whoami().get("id"))

# 2) Run a batch from a JSONL file
sim = Simulator(client, notify_promising=True, verbose=True)
results = sim.simulate_batch(
    "alphas.jsonl",
    save_promising=True,
    min_sharpe=1.0,
    min_fitness=0.9,
    max_concurrent=3,
    pause_every=20,
    pause_seconds=300,
)

# 3) Summarize
ok        = [r for r in results if r["ok"]]
promising = [r for r in results if r["promising"]]
errors    = [r for r in results if r["error"]]
print(f"\nRan {len(ok)} | promising {len(promising)} | errors {len(errors)}")

# 4) Review everything kept (across all runs)
for a in Simulator.load_promising():
    flip = "  [negate first]" if a["inverse"] else ""
    print(f"{a['alpha_id']}  sharpe={a['sharpe']}  fitness={a['fitness']}  {a['regular']}{flip}")

See examples/run_from_jsonl.py for a runnable version (with a manual-loop variant).

---

Testing

python tests/test_full_loop.py   # mock API — no network or credentials needed

Covers login, biometric Scan, single/batch simulation, the promising filter (incl. flip candidates), resume, error handling, auto re-login, and 429 backoff.

---

License

MIT