DeepSeek R1 reasoning + Purple Flea = cost-efficient financial agents

from openai import OpenAI
import requests, json, re

# DeepSeek is 100% OpenAI-compatible — just change the base URL
ds_client = OpenAI(
    base_url = "https://api.deepseek.com",
    api_key  = "sk-your-deepseek-key-here"
)
PF_API_KEY = "pf_live_your_key_here"
PF_BASE    = "https://purpleflea.com/api/v1"

def get_market_data(market: str) -> dict:
    """Fetch live market snapshot from Purple Flea."""
    resp = requests.get(
        f"{PF_BASE}/trading/market/{market}",
        headers={"X-API-Key": PF_API_KEY}
    )
    return resp.json()

def r1_trading_decision(market_data: dict) -> dict:
    """Use DeepSeek R1 to reason through a trading decision."""
    prompt = f"""Market: {market_data['market']}
Price: ${market_data['price']:,.2f} ({market_data['change_24h']:+.2f}% 24h)
Funding rate: {market_data['funding_rate']:.4f}%/8h
Open interest: ${market_data['open_interest']/1e9:.2f}B
Bid/ask spread: {market_data['spread_bps']:.1f} bps
Recent sentiment: {market_data.get('sentiment', 'neutral')}

Analyze this market. Reason through bullish and bearish factors.
Then output your decision as valid JSON only:
{{"action": "long"|"short"|"hold", "confidence": 0-100,
  "size_pct": 0-20, "leverage": 1-5,
  "stop_loss_pct": 1-10, "take_profit_pct": 5-30}}"""

    response = ds_client.chat.completions.create(
        model    = "deepseek-reasoner",  # R1 model
        messages = [
            {
                "role": "system",
                "content": (
                    "You are a quantitative crypto trading agent. "
                    "Always reason step by step before deciding. "
                    "Your reasoning chain is logged for auditability."
                )
            },
            {"role": "user", "content": prompt}
        ],
        temperature = 0.1
    )

    # R1 exposes reasoning_content separately from the final answer
    reasoning = response.choices[0].message.reasoning_content
    answer    = response.choices[0].message.content

    # Log the reasoning chain for audit trail
    print("[R1 Reasoning preview]", reasoning[:300], "...")

    json_match = re.search(r'\{[^{}]+\}', answer, re.DOTALL)
    if json_match:
        return json.loads(json_match.group())
    return {"action": "hold", "confidence": 0}

def execute_trade(decision: dict, market: str, balance: float) -> dict | None:
    """Execute Purple Flea perp trade from R1 decision."""
    if decision["action"] == "hold":
        print("  R1 decided: HOLD — no trade")
        return None
    if decision["confidence"] < 65:
        print(f"  Confidence {decision['confidence']}% too low — skip")
        return None

    size = balance * (decision["size_pct"] / 100)
    resp = requests.post(
        f"{PF_BASE}/trading/perp/open",
        headers={"X-API-Key": PF_API_KEY},
        json={
            "market":        market,
            "side":          decision["action"],
            "size_usd":      round(size, 2),
            "leverage":      decision.get("leverage", 2),
            "stop_loss_pct": decision.get("stop_loss_pct", 5),
            "order_type":    "market"
        }
    )
    return resp.json()

def run_agent(markets: list[str]):
    """Run one R1 reasoning cycle across multiple markets."""
    balance = requests.get(
        f"{PF_BASE}/wallet/balance",
        headers={"X-API-Key": PF_API_KEY}
    ).json()["usdc_balance"]

    print(f"[+] DeepSeek R1 agent starting | Balance: ${balance:.2f} USDC")

    for mkt in markets:
        print(f"\n[+] Analyzing {mkt} with DeepSeek R1...")
        data     = get_market_data(mkt)
        decision = r1_trading_decision(data)
        print(f"    Decision: {decision['action']} | Confidence: {decision['confidence']}%")
        result   = execute_trade(decision, mkt, balance)
        if result:
            print(f"    Executed: {result}")

if __name__ == "__main__":
    run_agent(["BTC-PERP", "ETH-PERP", "SOL-PERP"])