Scheduled Trading Agent
import modal
import purpleflea as pf
app = modal.App("purpleflea-trader")
image = modal.Image.debian_slim().pip_install("purpleflea")
pf_secret = modal.Secret.from_name("purpleflea-api-key")
@app.function(
image=image,
secrets=[pf_secret],
schedule=modal.Period(hours=8)
)
def funding_rate_harvest():
import os
client = pf.Client(api_key=os.environ["PURPLEFLEA_API_KEY"])
markets = client.trading.get_all_markets()
high_funding = [
m for m in markets
if m['funding_rate'] > 0.0005
]
for market in high_funding[:3]:
trade = client.trading.open_trade(
symbol=market['symbol'],
side="short",
size=50,
leverage=1
)
print(f"Opened: {market['symbol']} | funding: {market['funding_rate']:.4%}")
if __name__ == "__main__":
modal.runner.deploy_app(app)
Parallel Market Scanner
import modal
import purpleflea as pf
app = modal.App("market-scanner")
image = modal.Image.debian_slim().pip_install("purpleflea")
secret = modal.Secret.from_name("purpleflea-api-key")
@app.function(image=image, secrets=[secret])
def analyze_market(symbol: str) -> dict:
import os
client = pf.Client(api_key=os.environ["PURPLEFLEA_API_KEY"])
price = client.trading.get_price(symbol)
candles = client.trading.get_candles(symbol, "1h", limit=50)
closes = [c['close'] for c in candles]
momentum = (closes[-1] - closes[-10]) / closes[-10]
return {
"symbol": symbol,
"price": price['price'],
"momentum": momentum,
"signal": "long" if momentum > 0.02 else "short" if momentum < -0.02 else "neutral"
}
@app.local_entrypoint()
def main():
symbols = ["BTC", "ETH", "SOL", "BNB", "AVAX", "MATIC"]
results = list(analyze_market.map(symbols))
long_signals = [r for r in results if r['signal'] == 'long']
print(f"Long signals: {[r['symbol'] for r in long_signals]}")
Agent as Web Endpoint
import modal
from fastapi import FastAPI
import purpleflea as pf
web_app = FastAPI()
app = modal.App("purpleflea-agent-api")
image = modal.Image.debian_slim().pip_install("purpleflea", "fastapi")
secret = modal.Secret.from_name("purpleflea-api-key")
@web_app.post("/trade")
async def execute_trade(symbol: str, side: str, size: float):
import os
client = pf.Client(api_key=os.environ["PURPLEFLEA_API_KEY"])
result = client.trading.open_trade(
symbol=symbol, side=side, size=size
)
return result
@app.function(image=image, secrets=[secret])
@modal.asgi_app()
def fastapi_app():
return web_app
Casino Agent with Retries
import modal
import purpleflea as pf
app = modal.App("casino-agent")
image = modal.Image.debian_slim().pip_install("purpleflea")
secret = modal.Secret.from_name("purpleflea-api-key")
@app.function(
image=image,
secrets=[secret],
retries=modal.Retries(max_retries=3, backoff_coefficient=2.0)
)
def run_kelly_session(starting_balance: float, n_bets: int = 20) -> dict:
import os
client = pf.Client(api_key=os.environ["PURPLEFLEA_API_KEY"])
wins, losses = 0, 0
balance = starting_balance
for _ in range(n_bets):
bet = balance * 0.02
result = client.casino.play_dice(
bet_amount=str(bet),
target=49,
direction="over"
)
if result['won']:
wins += 1
balance += result['payout']
else:
losses += 1
balance -= bet
return {"wins": wins, "losses": losses, "final_balance": balance}