Social Signal Trading: How Agents Trade on Crypto Sentiment

import requests
import time
from dataclasses import dataclass
from typing import Optional, List
from datetime import datetime, timedelta

# Purple Flea config
PF_BASE   = "https://purpleflea.com"
PF_KEY    = "pf_live_your_key_here"
PF_HEADERS = {"Authorization": f"Bearer {PF_KEY}"}

# Sentiment API config (replace with your provider)
SENTIMENT_BASE = "https://api.sentiment-provider.com/v1"
SENTIMENT_KEY  = "your_sentiment_api_key"
SENT_HEADERS   = {"Authorization": f"Bearer {SENTIMENT_KEY}"}

# Monitored tokens
WATCHLIST = ["ETH", "SOL", "BTC", "MATIC", "LINK", "ARB", "OP"]

# Risk parameters
POSITION_SIZE_USDC = 500    # per trade
LEVERAGE           = 3
STOP_LOSS_PCT      = 0.025  # 2.5% stop
TAKE_PROFIT_PCT    = 0.075  # 7.5% target (3:1 R:R)


@dataclass
class SentimentSignal:
    token: str
    sentiment_score: float    # -1.0 to +1.0
    mention_volume: int
    mention_volume_vs_avg: float
    influencer_score: float   # weighted by track record
    sources: List[str]
    is_actionable: bool
    direction: str            # "long", "short", or "none"


def fetch_sentiment(token: str) -> SentimentSignal:
    """Fetch aggregated social sentiment for a token."""
    resp = requests.get(
        f"{SENTIMENT_BASE}/aggregate",
        params={
            "token": token,
            "sources": "twitter,farcaster,telegram,reddit",
            "window": "1h",
            "influencer_filter": "tracked_only"
        },
        headers=SENT_HEADERS
    ).json()

    score      = resp.get("sentiment_score", 0)
    volume     = resp.get("mention_count", 0)
    vol_ratio  = resp.get("vs_6h_avg", 1.0)
    inf_score  = resp.get("influencer_weighted_score", 0)
    sources    = resp.get("top_sources", [])

    # Determine if signal is actionable
    actionable = (
        vol_ratio >= 3.0            # at least 3x mention spike
        and abs(score) >= 0.4         # strong directional sentiment
        and abs(inf_score) >= 0.3     # influencers confirm direction
    )

    direction = "none"
    if actionable:
        direction = "long" if score > 0 else "short"

    return SentimentSignal(
        token=token,
        sentiment_score=score,
        mention_volume=volume,
        mention_volume_vs_avg=vol_ratio,
        influencer_score=inf_score,
        sources=sources,
        is_actionable=actionable,
        direction=direction
    )


def check_volume_confirmation(token: str) -> bool:
    """Confirm DEX volume spike via Purple Flea Analytics."""
    r = requests.get(
        f"{PF_BASE}/analytics/dex-volume",
        params={"pair": f"{token}/USDC", "chain": "ethereum"},
        headers=PF_HEADERS
    ).json()
    vol_ratio = r.get("volume_vs_30d_avg", 1.0)
    return vol_ratio >= 1.5


def check_price_momentum(token: str, direction: str) -> bool:
    """Confirm price is moving in the signal direction (not reversed)."""
    price_data = requests.get(
        f"{PF_BASE}/trading/price",
        params={"symbol": f"{token}-PERP", "interval": "15m", "periods": 4},
        headers=PF_HEADERS
    ).json()

    prices = price_data.get("closes", [])
    if len(prices) < 2:
        return False

    # Check 1h price change
    pct_change = (prices[-1] - prices[0]) / prices[0]

    # Avoid entering if price already moved 5%+ in signal direction
    if direction == "long" and pct_change > 0.05:
        return False   # too late
    if direction == "short" and pct_change < -0.05:
        return False  # too late

    # Confirm momentum is moving in signal direction
    if direction == "long":
        return pct_change > -0.01  # not in heavy downtrend
    else:
        return pct_change < 0.01   # not in heavy uptrend


def open_position(signal: SentimentSignal) -> Optional[dict]:
    """Open a perpetual position via Purple Flea Trading API."""
    resp = requests.post(
        f"{PF_BASE}/trading/perps/open",
        json={
            "symbol": f"{signal.token}-PERP",
            "side": signal.direction,
            "notional_usdc": POSITION_SIZE_USDC,
            "leverage": LEVERAGE,
            "stop_loss_pct": STOP_LOSS_PCT,
            "take_profit_pct": TAKE_PROFIT_PCT,
            "signal_source": "social_sentiment",
            "signal_score": signal.sentiment_score,
            "metadata": {
                "mention_spike": signal.mention_volume_vs_avg,
                "influencer_score": signal.influencer_score,
                "sources": signal.sources
            }
        },
        headers=PF_HEADERS
    ).json()

    if resp.get("order_id"):
        print(f"Opened {signal.direction} {signal.token} | "
              f"Sentiment: {signal.sentiment_score:+.2f} | "
              f"Spike: {signal.mention_volume_vs_avg:.1f}x | "
              f"Order: {resp['order_id']}")
    return resp


# Main agent loop
if __name__ == "__main__":
    print("Social Signal Trading Agent started")
    open_positions = {}

    while True:
        for token in WATCHLIST:
            # Skip if already in a position for this token
            if token in open_positions:
                continue

            signal = fetch_sentiment(token)

            if not signal.is_actionable:
                continue

            # Filter 1: Volume confirmation
            if not check_volume_confirmation(token):
                print(f"{token}: Signal blocked — no volume confirmation")
                continue

            # Filter 2: Price momentum
            if not check_price_momentum(token, signal.direction):
                print(f"{token}: Signal blocked — price already moved")
                continue

            # All filters passed — open position
            result = open_position(signal)
            if result and result.get("order_id"):
                open_positions[token] = result["order_id"]

        time.sleep(300)  # Scan every 5 minutes