Python SDK

pip install propdao

export PROPDAO_API_KEY=pd_live_...
export PROPDAO_ACCOUNT=prop-abc123   # optional; examples default to your first account

"""PropDAO agent SDK. Zero dependencies.

    from propdao import PropDAO
    pd = PropDAO()                      # reads PROPDAO_API_KEY
    acct = pd.get_accounts()[0]["account_id"]
    print(pd.get_risk(acct)["roomUsd"])  # dollars you can lose before breach
    pd.market_buy(acct, "BTCUSDC", 0.01, leverage=2, sl=60000)
"""
import json
import os
import secrets
import time
import urllib.error
import urllib.request

__all__ = ["PropDAO", "PropDAOError"]
__version__ = "0.1.0"


class PropDAOError(Exception):
    def __init__(self, status, message):
        super().__init__(f"{status}: {message}")
        self.status = status
        self.message = message


class PropDAO:
    def __init__(self, api_key=None, base_url="https://app.propdao.finance/api/v1", timeout=15):
        self.api_key = api_key or os.environ.get("PROPDAO_API_KEY")
        if not self.api_key:
            raise ValueError("api_key required (or set PROPDAO_API_KEY)")
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def _req(self, method, path, body=None, _retried=False):
        data = None if body is None else json.dumps(body).encode()
        req = urllib.request.Request(
            self.base_url + path, data=data, method=method,
            headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json",
                     "User-Agent": f"propdao-python/{__version__}"},
        )
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
                return json.load(r)
        except urllib.error.HTTPError as e:
            # 429 on a read: wait out Retry-After (capped) once, then surface it.
            if e.code == 429 and method == "GET" and not _retried:
                time.sleep(min(5, float(e.headers.get("Retry-After") or 1)))
                return self._req(method, path, body, _retried=True)
            try:
                msg = json.load(e).get("error", e.reason)
            except Exception:
                msg = e.reason
            raise PropDAOError(e.code, msg) from None
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            # status 0 = never got a response; the intentId makes a retry safe.
            raise PropDAOError(0, f"network error: {getattr(e, 'reason', e)}") from None

    # -- read --------------------------------------------------------------
    def health(self):
        return self._req("GET", "/health")

    def markets(self):
        """Every tradable symbol with maxLeverage, lotStep, and the fee schedule. No key needed."""
        return self._req("GET", "/markets")

    def challenges(self):
        """Account sizes, prices and rules. No key needed."""
        return self._req("GET", "/challenges")["data"]

    def me(self):
        return self._req("GET", "/me")

    def get_accounts(self):
        return self._req("GET", "/accounts").get("accounts", [])

    def get_account(self, account_id):
        return self._req("GET", f"/accounts/{account_id}")

    def get_open_positions(self, account_id):
        return self._req("GET", f"/accounts/{account_id}/positions?limit=500")["data"]

    def get_open_orders(self, account_id):
        return self._req("GET", f"/accounts/{account_id}/orders?limit=500")["data"]

    def get_trades(self, account_id, limit=100, offset=0):
        return self._req("GET", f"/accounts/{account_id}/trades?limit={limit}&offset={offset}")["data"]

    def get_risk(self, account_id):
        """equity, floor, roomUsd, roomPct, breached — size positions off roomUsd."""
        return self._req("GET", f"/accounts/{account_id}/risk")

    # -- trade -------------------------------------------------------------
    def place_order(self, account_id, symbol, side, qty, order_type="market", leverage=None,
                    limit_price=None, trigger_price=None, sl=None, tp=None, intent_id=None, **extra):
        """order_type: market | limit | stop_market | stop_limit | take_market | take_limit | scale | twap.
        intent_id: idempotency key (auto-generated if omitted, so a network retry never double-fills).
        extra: twapMs, twapMin, twapMax, twapTrigger, scaleStart, scaleEnd, scaleCount, ..."""
        body = {"symbol": symbol, "side": side.upper(), "qty": qty, "orderType": order_type, "leverage": leverage,
                "limitPrice": limit_price, "triggerPrice": trigger_price, "sl": sl, "tp": tp,
                "intentId": intent_id or secrets.token_urlsafe(16), **extra}
        body = {k: v for k, v in body.items() if v is not None}
        for attempt in range(5):
            try:
                return self._req("POST", f"/accounts/{account_id}/orders", body)
            except PropDAOError as e:
                # 409 = same intentId still executing (a retried request raced itself); wait for its result.
                if e.status != 409 or attempt == 4:
                    raise
                time.sleep(0.5 * (attempt + 1))

    def market_buy(self, account_id, symbol, qty, **kw):
        return self.place_order(account_id, symbol, "BUY", qty, **kw)

    def market_sell(self, account_id, symbol, qty, **kw):
        return self.place_order(account_id, symbol, "SELL", qty, **kw)

    def limit_buy(self, account_id, symbol, qty, price, **kw):
        return self.place_order(account_id, symbol, "BUY", qty, order_type="limit", limit_price=price, **kw)

    def limit_sell(self, account_id, symbol, qty, price, **kw):
        return self.place_order(account_id, symbol, "SELL", qty, order_type="limit", limit_price=price, **kw)

    def twap(self, account_id, symbol, side, qty, minutes, leverage=None, min_price=None, max_price=None, **kw):
        """Work `qty` into the market over `minutes`. Returns the account state; the TWAP appears in twapOrders."""
        return self.place_order(account_id, symbol, side, qty, order_type="twap", leverage=leverage,
                                twapMs=int(minutes * 60_000), twapMin=min_price, twapMax=max_price, **kw)

    def set_risk(self, account_id, position_id, sl=None, tp=None):
        """Set or clear stop-loss / take-profit on an open position (None leaves it unchanged, pass 0 to clear)."""
        body = {}
        if sl is not None: body["sl"] = None if sl == 0 else sl
        if tp is not None: body["tp"] = None if tp == 0 else tp
        return self._req("PATCH", f"/accounts/{account_id}/positions/{position_id}", body)

    def cancel_order(self, account_id, order_id):
        return self._req("DELETE", f"/accounts/{account_id}/orders/{order_id}")

    def cancel_all_orders(self, account_id):
        """Cancel every resting order and TWAP. Cancelling one rung of a scale ladder
        cancels its siblings, so a 404 on a later rung is expected and skipped."""
        out = []
        for o in self.get_open_orders(account_id):
            try:
                if o.get("orderType") == "twap":
                    out.append(self.cancel_twap(account_id, o["id"]))
                else:
                    out.append(self.cancel_order(account_id, o["id"]))
            except PropDAOError as e:
                if e.status != 404:
                    raise
        return out

    def cancel_twap(self, account_id, twap_id):
        return self._req("DELETE", f"/accounts/{account_id}/twaps/{twap_id}")

    def close_position(self, account_id, position_id, percent=1.0):
        return self._req("POST", f"/accounts/{account_id}/positions/{position_id}/close", {"percent": percent})

    def close_all(self, account_id):
        return [self.close_position(account_id, p["id"]) for p in self.get_open_positions(account_id)]

from propdao import PropDAO

pd = PropDAO()                                   # reads PROPDAO_API_KEY
acct = pd.get_accounts()[0]["account_id"]        # first account you own

risk = pd.get_risk(acct)                         # ALWAYS before sizing
print(f"room to breach: ${risk['roomUsd']:.2f} ({risk['floorKind']} floor)")

r = pd.market_buy(acct, "BTCUSDC", 0.01, leverage=2, sl=60000)
print(r["status"])                               # MARKET BUY executed @ 64210.50 (BTCUSDC, 2x). Fee -$0.10.

pd = PropDAO(
    api_key="pd_live_...",                             # or PROPDAO_API_KEY env var
    base_url="https://app.propdao.finance/api/v1",     # default
    timeout=15,                                        # seconds
)

from propdao import PropDAO

pd = PropDAO(); acct = pd.get_accounts()[0]["account_id"]
SYMBOL, STOP_PCT, RISK_FRACTION = "ETHUSDC", 0.02, 0.25

risk = pd.get_risk(acct)
mkt = next(m for m in pd.markets()["data"] if m["symbol"] == SYMBOL)
px = next((p["mark"] for p in pd.get_open_positions(acct) if p["symbol"] == SYMBOL), None) or 2600.0

budget = risk["roomUsd"] * RISK_FRACTION          # dollars we are willing to lose on this trade
qty = budget / (px * STOP_PCT)                    # units such that a 2% adverse move loses `budget`
qty = round(qty / mkt["lotStep"]) * mkt["lotStep"]

r = pd.market_buy(acct, SYMBOL, qty, leverage=min(2, mkt["maxLeverage"]), sl=px * (1 - STOP_PCT))
print(r["status"])

from propdao import PropDAO

pd = PropDAO(); acct = pd.get_accounts()[0]["account_id"]
SYMBOL, MID, STEP, QTY = "BTCUSDC", 81000.0, 500.0, 0.001

for i in range(1, 6):
    pd.limit_buy(acct, SYMBOL, QTY, MID - STEP * i, leverage=2)
    pd.limit_sell(acct, SYMBOL, QTY, MID + STEP * i, leverage=2)

print(len(pd.get_open_orders(acct)), "orders resting")

"""DCA bot: buy a fixed dollar amount of BTC every N minutes, but only while
the account has room. Stops itself when room gets thin.

    PROPDAO_API_KEY=pd_live_... python dca_bot.py
"""
import os
import time
from propdao import PropDAO

ACCOUNT = os.environ.get("PROPDAO_ACCOUNT")     # optional; defaults to first account
SYMBOL = "BTCUSDC"
USD_PER_BUY = 200
EVERY_MIN = 60
MIN_ROOM_USD = 500                              # stop buying below this

pd = PropDAO()
acct = ACCOUNT or pd.get_accounts()[0]["account_id"]
lot = next(m["lotStep"] for m in pd.markets()["data"] if m["symbol"] == SYMBOL)

while True:
    risk = pd.get_risk(acct)
    if risk["breached"]:
        print("account breached, stopping"); break
    if risk["roomUsd"] < MIN_ROOM_USD:
        print(f"room ${risk['roomUsd']:.0f} < ${MIN_ROOM_USD}, skipping this round")
    else:
        # price from the position mark if we have one, else from the latest trade
        state = pd.get_account(acct)
        px = next((p["mark"] for p in state.get("openPositions", []) if p["symbol"] == SYMBOL and p.get("mark")), None)
        qty = round((USD_PER_BUY / px) / lot) * lot if px else lot
        r = pd.market_buy(acct, SYMBOL, qty, leverage=1)
        print(r.get("status"))
    time.sleep(EVERY_MIN * 60)

"""Risk guard: run alongside any strategy. Every few seconds, if room to
breach drops below a floor you choose, flatten everything and cancel all
working orders — before the platform does it for you at the floor.

    PROPDAO_API_KEY=pd_live_... python risk_guard.py
"""
import os
import time
from propdao import PropDAO

ACCOUNT = os.environ.get("PROPDAO_ACCOUNT")
MIN_ROOM_PCT = 1.0          # flatten when < 1% of equity remains above the floor
POLL_S = 3

pd = PropDAO()
acct = ACCOUNT or pd.get_accounts()[0]["account_id"]

while True:
    r = pd.get_risk(acct)
    print(f"equity {r['equity']:.2f}  floor {r['floor']:.2f}  room {r['roomUsd']:.2f} ({r['roomPct']:.2f}%)")
    if r["openPositions"] and r["roomPct"] < MIN_ROOM_PCT:
        print("room too thin — flattening")
        pd.cancel_all_orders(acct)
        pd.close_all(acct)
    time.sleep(POLL_S)

r = pd.market_buy(acct, "ETHUSDC", 0.1, leverage=2, sl=2500, tp=2900)

# ...or adjust on the open position
pos = pd.get_open_positions(acct)[0]
pd.set_risk(acct, pos["id"], sl=2550)          # move the stop up
pd.set_risk(acct, pos["id"], tp=0)             # 0 clears the take-profit

# work 0.5 BTC in over 30 minutes, skipping slices above 82,000
state = pd.twap(acct, "BTCUSDC", "BUY", 0.5, minutes=30, leverage=2, max_price=82000)
twap_id = state["twapOrders"][-1]["id"]
# ...later
pd.cancel_twap(acct, twap_id)

pd = PropDAO(); acct = pd.get_accounts()[0]["account_id"]
r = pd.get_risk(acct)
print(f"equity ${r['equity']:,.2f}   floor ${r['floor']:,.2f}   room ${r['roomUsd']:,.2f} ({r['roomPct']:.2f}%)")
print(f"{'Symbol':<10}{'Side':<6}{'Qty':>10}{'Entry':>12}{'Mark':>12}{'uPnL':>10}")
for p in pd.get_open_positions(acct):
    print(f"{p['symbol']:<10}{p['side']:<6}{p['qty']:>10}{p['entry']:>12,.2f}{p['mark']:>12,.2f}{p['unrealizedPnl']:>10,.2f}")

"""An LLM trading agent in ~60 lines: Claude gets the PropDAO SDK as tools,
checks its room to breach, and decides. Run it on a schedule (cron, every
15 minutes) rather than in a tight loop — the model, not a timer, sets the pace.

    pip install propdao anthropic
    PROPDAO_API_KEY=pd_live_... ANTHROPIC_API_KEY=sk-ant-... python llm_agent.py
"""
import json
import os

import anthropic
from anthropic import beta_tool
from propdao import PropDAO

pd = PropDAO()
ACCT = os.environ.get("PROPDAO_ACCOUNT") or pd.get_accounts()[0]["account_id"]


@beta_tool
def get_risk() -> str:
    """Equity, drawdown floor and roomUsd (dollars that can be lost before the account is closed). Check before every trade."""
    return json.dumps(pd.get_risk(ACCT))


@beta_tool
def get_positions() -> str:
    """Open positions with entry, qty, side, leverage, sl, tp."""
    return json.dumps(pd.get_open_positions(ACCT))


@beta_tool
def place_order(symbol: str, side: str, qty: float, leverage: float = 1, sl: float | None = None, tp: float | None = None) -> str:
    """Market order. symbol like BTCUSDC; side BUY or SELL; qty in units of the asset.

    Args:
        symbol: e.g. BTCUSDC, ETHUSDC
        side: BUY or SELL
        qty: units of the asset, not dollars
        leverage: 1-5 depending on symbol
        sl: stop-loss price
        tp: take-profit price
    """
    r = pd.place_order(ACCT, symbol, side, qty, leverage=leverage, sl=sl, tp=tp)
    return r.get("status", "ok")


@beta_tool
def close_position(position_id: str) -> str:
    """Close an open position at market."""
    return json.dumps(pd.close_position(ACCT, position_id).get("status", "ok"))


client = anthropic.Anthropic()
runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=16000,
    system=(
        "You manage one PropDAO prop-firm account. The account is closed permanently the moment "
        "equity touches the drawdown floor. Rules: always call get_risk first; never risk more than "
        "a quarter of roomUsd on any single position's stop; prefer doing nothing over a marginal trade. "
        "Finish with one line: what you did and why."
    ),
    tools=[get_risk, get_positions, place_order, close_position],
    messages=[{"role": "user", "content": "Review the account and act if there is a clear, well-sized opportunity in BTC or ETH."}],
)
final = runner.until_done()
for block in final.content:
    if block.type == "text":
        print(block.text)

from propdao import PropDAO, PropDAOError

try:
    pd.market_buy(acct, "BTCUSDC", 0.01, leverage=2)
except PropDAOError as e:
    if e.status == 429:
        time.sleep(2)                       # rate limited
    elif e.status == 403 and "breached" in e.message:
        raise SystemExit("account is over")
    elif e.status == 0:
        pass                                # network blip - retry with the same intent_id
    elif e.status == 400 and e.message.startswith("Hold positions"):
        time.sleep(1)                       # 1 s min hold / 0.5 s between executions
    else:
        print("API error", e.status, e.message)