traderspost/traderspost-command-dash/traderspost-webhook

https://webhooks.traderspost.io/trading/webhook/<strategy-id>

{ "action": "buy",  "ticker": "AAPL" }
{ "action": "sell", "ticker": "AAPL" }
{ "action": "exit", "ticker": "AAPL" }

{
  "action": "buy",
  "ticker": "AAPL",
  "quantity": 10,             // shares — omit to use strategy's default sizing
  "price": 234.56,            // limit price — omit for market
  "sentiment": "bullish",     // freeform; surfaces in TradersPost UI
  "signalPrice": 234.50,      // price at which the signal was generated
  "stopLoss": { "type": "stop_loss", "stopPrice": 230.00 },
  "takeProfit": { "limitPrice": 240.00 }
}

{ "action": "buy", "ticker": "BTC.USD" }

import httpx
async def post_webhook(url: str, payload: dict) -> dict:
    """Returns {ok, status, body}. Raises on transport errors only."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.post(url, json=payload)
        return {"ok": r.is_success, "status": r.status_code, "body": r.text[:500]}
def buy(ticker: str, quantity: float | None = None) -> dict:
    out = {"action": "buy", "ticker": ticker}
    if quantity is not None:
        out["quantity"] = quantity
    return out
def sell(ticker: str, quantity: float | None = None) -> dict:
    out = {"action": "sell", "ticker": ticker}
    if quantity is not None:
        out["quantity"] = quantity
    return out
def exit_(ticker: str) -> dict:
    return {"action": "exit", "ticker": ticker}

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
import logging
log = logging.getLogger("scheduler")
scheduler = AsyncIOScheduler()
scheduler.start()
_jobs: dict[str, str] = {}  # strategy_name → job_id
async def _tick(name: str, webhook: str, signal_fn) -> None:
    payload = signal_fn()  # your strategy returns a dict or None
    if not payload:
        return
    result = await post_webhook(webhook, payload)
    log.info("%s → %s", name, result)
def start_live(name: str, cron: str, webhook: str, signal_fn) -> str:
    """`cron` follows standard cron syntax — '*/5 9-15 * * 1-5' = every 5 min, weekdays, market hours."""
    job = scheduler.add_job(
        _tick, CronTrigger.from_crontab(cron),
        kwargs={"name": name, "webhook": webhook, "signal_fn": signal_fn},
        id=f"strategy-{name}", replace_existing=True,
    )
    _jobs[name] = job.id
    return job.id
def stop_live(name: str) -> bool:
    job_id = _jobs.pop(name, None)
    if job_id:
        scheduler.remove_job(job_id)
        return True
    return False