Redis-Powered AI Agents
That Trade

import redis
import json
import time
import httpx
from typing import Optional, Dict, Any

class RedisAgentCache:
    """
    Read-through cache for Purple Flea API calls.
    All hot data lives in Redis; cold paths hit the API.
    """

    def __init__(self, redis_url: str, api_key: str, agent_id: str):
        self.r = redis.from_url(redis_url, decode_responses=True)
        self.api_key = api_key            # e.g. pf_live_yourkey
        self.agent_id = agent_id
        self.base = "https://purpleflea.com/api"
        self.http = httpx.Client(
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10
        )

    # ── Price Cache ──────────────────────────────────────────────────────────

    def get_price(self, market: str, ttl: int = 1) -> Dict[str, Any]:
        """
        Return cached price for `market` (e.g. 'BTC-PERP').
        Fetches from Purple Flea trading API if cache miss.
        TTL=1s by default — aligns with tick frequency.
        """
        cache_key = f"price:{market}"
        cached = self.r.get(cache_key)

        if cached:
            data = json.loads(cached)
            data["_cache"] = "hit"
            return data

        # Cache miss — fetch from Purple Flea
        resp = self.http.get(f"{self.base}/trading/markets/{market}/ticker")
        resp.raise_for_status()
        data = resp.json()

        self.r.setex(cache_key, ttl, json.dumps(data))
        data["_cache"] = "miss"
        return data

    # ── Balance Cache ─────────────────────────────────────────────────────────

    def cache_balance(self, wallet_address: str, ttl: int = 30) -> Dict[str, Any]:
        """
        Cache wallet balance. TTL=30s balances freshness against API rate limits.
        Uses hash for O(1) field reads without deserializing the full object.
        """
        cache_key = f"balance:{wallet_address}"
        cached_hash = self.r.hgetall(cache_key)

        if cached_hash:
            return {
                "address": wallet_address,
                "usdc": float(cached_hash["usdc"]),
                "btc": float(cached_hash["btc"]),
                "eth": float(cached_hash["eth"]),
                "cached_at": cached_hash["cached_at"],
                "_cache": "hit",
            }

        resp = self.http.get(f"{self.base}/wallet/balance/{wallet_address}")
        resp.raise_for_status()
        data = resp.json()

        pipe = self.r.pipeline()
        pipe.hset(cache_key, mapping={
            "usdc": str(data["usdc"]),
            "btc": str(data.get("btc", 0)),
            "eth": str(data.get("eth", 0)),
            "cached_at": str(time.time()),
        })
        pipe.expire(cache_key, ttl)
        pipe.execute()

        data["_cache"] = "miss"
        return data

    # ── Position State ────────────────────────────────────────────────────────

    def set_position(self, market: str, side: str, size: float,
                      entry_price: float, ttl: int = 3600):
        """Store open position state. Expires after 1h (session-length TTL)."""
        key = f"position:{self.agent_id}:{market}"
        self.r.hset(key, mapping={
            "side": side,
            "size": str(size),
            "entry_price": str(entry_price),
            "opened_at": str(time.time()),
        })
        self.r.expire(key, ttl)

    def get_position(self, market: str) -> Optional[Dict]:
        key = f"position:{self.agent_id}:{market}"
        data = self.r.hgetall(key)
        return data if data else None

    def clear_position(self, market: str):
        self.r.delete(f"position:{self.agent_id}:{market}")

import redis
import time

r = redis.from_url("redis://localhost:6379")

# Lua script — runs atomically on Redis server (no race conditions)
TOKEN_BUCKET_SCRIPT = """
local key        = KEYS[1]
local capacity   = tonumber(ARGV[1])   -- max tokens
local refill_rate = tonumber(ARGV[2])  -- tokens per second
local now        = tonumber(ARGV[3])   -- current time (ms)
local requested  = tonumber(ARGV[4])   -- tokens needed

local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now

-- Refill based on elapsed time
local elapsed = (now - last_refill) / 1000
local new_tokens = math.min(capacity, tokens + elapsed * refill_rate)

if new_tokens >= requested then
    new_tokens = new_tokens - requested
    redis.call('HMSET', key, 'tokens', new_tokens, 'last_refill', now)
    redis.call('EXPIRE', key, 3600)
    return 1    -- allowed
else
    redis.call('HMSET', key, 'tokens', new_tokens, 'last_refill', now)
    redis.call('EXPIRE', key, 3600)
    return 0    -- denied
end
"""

bucket_script = r.register_script(TOKEN_BUCKET_SCRIPT)

def acquire_token(api_key_hash: str, capacity: int = 60,
                   refill_rate: float = 1.0) -> bool:
    """
    Try to consume one API token.
    capacity=60: burst up to 60 calls
    refill_rate=1.0: refill 1 token/second (60 RPM steady state)
    Returns True if the call is allowed.
    """
    key = f"rate_limit:{api_key_hash}"
    now_ms = int(time.time() * 1000)
    result = bucket_script(
        keys=[key],
        args=[capacity, refill_rate, now_ms, 1]
    )
    return bool(result)

# Usage across all agents sharing one API key
def safe_api_call(endpoint: str, payload: dict, api_key: str):
    key_hash = api_key[-12:]          # last 12 chars as bucket key
    if not acquire_token(key_hash):
        raise Exception("Rate limit: wait for token refill")
    # ... proceed with API call

import redis, time, json

r = redis.from_url("redis://localhost:6379")

STREAM_KEY = "pf:audit"
MAX_STREAM_LEN = 100_000   # ~1M events before trimming oldest

def log_transaction(
    agent_id: str,
    action: str,        # "bet" | "withdraw" | "escrow_deposit" | "order"
    market: str,
    amount: float,
    result: dict,       # API response from Purple Flea
    metadata: dict = None
) -> str:
    """
    Append transaction to Redis Stream.
    Returns the stream entry ID (timestamp-sequence, e.g. '1709812345678-0').
    Use MAXLEN to cap stream size automatically.
    """
    entry = {
        "agent_id": agent_id,
        "action": action,
        "market": market,
        "amount": str(amount),
        "status": result.get("status", "unknown"),
        "tx_id": result.get("tx_id", ""),
        "pnl": str(result.get("pnl", 0)),
        "wall_time": str(time.time()),
        "meta": json.dumps(metadata or {}),
    }
    entry_id = r.xadd(STREAM_KEY, entry, maxlen=MAX_STREAM_LEN, approximate=True)
    return entry_id.decode()

def replay_transactions(
    agent_id: str = None,
    start: str = "0",       # stream ID or "-" for beginning
    count: int = 500
) -> list:
    """Read audit log entries, optionally filtering by agent."""
    entries = r.xrange(STREAM_KEY, min=start, count=count)
    results = []
    for entry_id, fields in entries:
        if agent_id and fields.get("agent_id") != agent_id:
            continue
        results.append({"id": entry_id, **fields})
    return results

def get_agent_pnl_from_stream(agent_id: str) -> float:
    """Calculate total P&L by replaying the agent's stream entries."""
    entries = replay_transactions(agent_id=agent_id, count=10_000)
    return sum(float(e["pnl"]) for e in entries)

# Example: log a Purple Flea casino bet
entry_id = log_transaction(
    agent_id="agent-007",
    action="bet",
    market="coin-flip",
    amount=0.10,
    result={"status": "win", "pnl": 0.09, "tx_id": "cf-9284"},
    metadata={"prediction": "heads", "model": "claude-opus-4"}
)
print(f"Logged: {entry_id}")