Social Media Intelligence

import pandas as pd
import numpy as np

def compute_buzz_metrics(df: pd.DataFrame, window: str = "1H") -> pd.DataFrame:
    """Compute time-series discussion-buzz metrics.

    Args:
        df: Message DataFrame containing timestamp / platform / ticker columns
        window: Aggregation window, such as "1H" / "4H" / "1D"

    Returns:
        Time-series DataFrame with:
        - msg_count: message volume
        - unique_authors: count of distinct active users
        - topic_freq: topic frequency as ticker volume / total message volume
        - engagement_score: weighted interaction count
        - buzz_zscore: message-volume Z-score for anomaly detection
    """
    df["timestamp"] = pd.to_datetime(df["timestamp"])
    df = df.set_index("timestamp").sort_index()

    result = df.resample(window).agg(
        msg_count=("text", "count"),
        unique_authors=("author_id", "nunique"),
        total_engagement=("engagement", "sum"),
    )

    # Topic frequency, i.e. relative buzz.
    total = df.resample(window)["text"].count()
    result["topic_freq"] = result["msg_count"] / total.replace(0, np.nan)

    # Z-score anomaly detection using a 30-window rolling baseline.
    roll_mean = result["msg_count"].rolling(30, min_periods=5).mean()
    roll_std = result["msg_count"].rolling(30, min_periods=5).std()
    result["buzz_zscore"] = (result["msg_count"] - roll_mean) / roll_std.replace(0, np.nan)

    return result

def compute_fear_greed_index(
    sentiment_series: pd.Series,
    buzz_series: pd.Series,
    lookback: int = 30,
) -> pd.Series:
    """Construct a CNN-style fear-and-greed index.

    Args:
        sentiment_series: Daily average sentiment in [-1, 1]
        buzz_series: Daily message-volume series
        lookback: Historical window in days used for percentile ranking

    Returns:
        Fear-and-greed index in [0, 100]
        0-20: extreme fear
        20-40: fear
        40-60: neutral
        60-80: greed
        80-100: extreme greed

    Note:
        Extreme greed (>80) is often a short-term top signal.
        Extreme fear (<20) is often a short-term bottom signal.
    """
    # Normalize sentiment by percentile rank.
    sentiment_rank = sentiment_series.rolling(lookback).rank(pct=True) * 100

    # Normalize buzz by percentile rank.
    buzz_rank = buzz_series.rolling(lookback).rank(pct=True) * 100

    # Weighted combination: 60% sentiment + 40% buzz.
    fear_greed = 0.6 * sentiment_rank + 0.4 * buzz_rank

    return fear_greed.clip(0, 100)

def classify_author_type(author: dict) -> str:
    """Classify an account as retail / institutional / KOL using profile traits.

    Args:
        author: Dict containing followers_count / verified / account_age_days /
                tweet_count / following_count

    Returns:
        'institutional': institutional account
        'kol': high-impact KOL
        'retail': retail user
        'bot_risk': suspected bot

    Note:
        Institutional sentiment should carry more weight than KOL,
        and KOL more than retail. Retail extremes often have contrarian value.
    """
    followers = author.get("followers_count", 0)
    verified = author.get("verified", False)
    age_days = author.get("account_age_days", 0)
    tweet_count = author.get("tweet_count", 0)

    # Bot-risk detection
    if age_days < 30 and tweet_count > 1000:
        return "bot_risk"
    if tweet_count > 0 and (tweet_count / max(age_days, 1)) > 50:
        return "bot_risk"

    # Institutional characteristics: verified plus large audience
    if verified and followers > 100_000:
        return "institutional"

    # KOL: large following but not necessarily verified
    if followers > 10_000:
        return "kol"

    return "retail"


def weighted_sentiment(
    df: pd.DataFrame,
    weights: dict | None = None,
) -> pd.Series:
    """Compute weighted sentiment by author category.

    Args:
        df: DataFrame containing sentiment_score / author_type / timestamp
        weights: Optional category weights, defaults to institutional:3, kol:2, retail:1

    Returns:
        Daily weighted sentiment time series
    """
    if weights is None:
        weights = {"institutional": 3.0, "kol": 2.0, "retail": 1.0, "bot_risk": 0.0}

    df["weight"] = df["author_type"].map(weights).fillna(1.0)
    df["weighted_sentiment"] = df["sentiment_score"] * df["weight"]

    return (
        df.groupby(df["timestamp"].dt.date)
        .apply(lambda g: g["weighted_sentiment"].sum() / g["weight"].sum())
        .rename("weighted_sentiment")
    )

import pandas as pd
import numpy as np
from scipy.stats import spearmanr

def compute_ic(
    factor_series: pd.Series,
    forward_return: pd.Series,
    method: str = "spearman",
) -> float:
    """Compute one-period factor IC (information coefficient).

    Args:
        factor_series: Cross-sectional factor values, e.g. same-day sentiment by ticker
        forward_return: Matching forward N-day returns
        method: "spearman" for rank correlation or "pearson"

    Returns:
        IC in [-1, 1]. |IC| > 0.05 is useful and > 0.1 is strong.
    """
    aligned = pd.concat([factor_series, forward_return], axis=1).dropna()
    if len(aligned) < 5:
        return np.nan

    if method == "spearman":
        ic, _ = spearmanr(aligned.iloc[:, 0], aligned.iloc[:, 1])
    else:
        ic = aligned.iloc[:, 0].corr(aligned.iloc[:, 1])
    return ic


def compute_icir(ic_series: pd.Series) -> float:
    """Compute ICIR, the information ratio of IC.

    Args:
        ic_series: Time series of IC values

    Returns:
        ICIR = mean(IC) / std(IC), where > 0.5 is useful and > 1.0 is strong
    """
    return ic_series.mean() / ic_series.std() if ic_series.std() > 0 else np.nan


# Example factor-construction workflow
def build_sentiment_factor(
    raw_data: pd.DataFrame,
    forward_days: int = 5,
) -> dict:
    """Build a complete social-sentiment factor and test its effectiveness.

    Args:
        raw_data: Raw data containing date / ticker / sentiment_score / author_type
        forward_days: Forward return horizon in days

    Returns:
        {'factor': DataFrame, 'ic_series': Series, 'icir': float}
    """
    # 1. Cross-sectional standardization
    factor = (
        raw_data.groupby("date")["sentiment_score"]
        .transform(lambda x: (x - x.mean()) / (x.std() + 1e-8))
    )
    factor_data = raw_data.assign(sentiment_norm=factor)

    # 2. Compute period-by-period IC
    ic_list = []
    dates = np.sort(raw_data["date"].unique())
    for date in dates[:-forward_days]:
        fwd_dates = dates[dates > date]
        if len(fwd_dates) < forward_days:
            continue
        fwd_date = fwd_dates[forward_days - 1]
        f = factor_data[factor_data["date"] == date].set_index("ticker")["sentiment_norm"]
        r = raw_data[raw_data["date"] == fwd_date].set_index("ticker")["return"]
        ic_list.append((date, compute_ic(f, r)))

    ic_series = pd.Series(dict(ic_list))
    return {
        "factor": factor,
        "ic_series": ic_series,
        "ic_mean": ic_series.mean(),
        "icir": compute_icir(ic_series),
    }

def orthogonalize_sentiment(
    sentiment_factor: pd.Series,
    traditional_factors: pd.DataFrame,
) -> pd.Series:
    """Orthogonalize the sentiment factor against traditional factors.

    Args:
        sentiment_factor: Raw sentiment factor after cross-sectional normalization
        traditional_factors: Matrix of traditional factors such as size / momentum / valuation

    Returns:
        Pure sentiment factor after removing shared components, i.e. the residual

    Note:
        Orthogonalization often lowers IC, but improves factor independence
        and reduces double-counting in multi-factor portfolios.
    """
    from sklearn.linear_model import LinearRegression
    import numpy as np

    # Regress on complete observations and keep missing rows missing.
    factors = traditional_factors.reindex(sentiment_factor.index)
    valid = sentiment_factor.notna() & factors.notna().all(axis=1)
    residual = pd.Series(np.nan, index=sentiment_factor.index, name="sentiment_orthogonal")
    if not valid.any():
        return residual

    X = factors.loc[valid].values
    y = sentiment_factor.loc[valid].values

    reg = LinearRegression(fit_intercept=True).fit(X, y)
    residual.loc[valid] = y - reg.predict(X)

    return residual

PLATFORM_WEIGHTS = {
    # Weight basis: historical IC contribution + information quality
    "twitter_institutional": 0.35,
    "twitter_kol": 0.20,
    "telegram_signal": 0.15,
    "telegram_research": 0.15,
    "discord_community": 0.10,
    "reddit_wsb": 0.05,
}

def aggregate_platform_sentiment(platform_scores: dict[str, float]) -> float:
    """Aggregate sentiment scores from multiple platforms using weights.

    Args:
        platform_scores: Dict of {platform_key: sentiment_score} with scores in [-1, 1]

    Returns:
        Aggregated sentiment score in [-1, 1]
    """
    total_weight = 0.0
    weighted_sum = 0.0

    for platform, score in platform_scores.items():
        weight = PLATFORM_WEIGHTS.get(platform, 0.05)
        weighted_sum += score * weight
        total_weight += weight

    return weighted_sum / total_weight if total_weight > 0 else 0.0

def detect_sentiment_reversal(
    fg_index: pd.Series,
    price_series: pd.Series,
    extreme_threshold: float = 80.0,
    fear_threshold: float = 20.0,
    confirmation_days: int = 3,
) -> pd.DataFrame:
    """Detect reversal signals from sentiment extremes.

    Args:
        fg_index: Fear-and-greed index in [0, 100]
        price_series: Corresponding price series
        extreme_threshold: Extreme-greed threshold, default 80
        fear_threshold: Extreme-fear threshold, default 20
        confirmation_days: Number of days required for confirmation

    Returns:
        DataFrame containing signal / direction / strength
        signal = 1: short signal due to sustained extreme greed
        signal = -1: long signal due to sustained extreme fear
        signal = 0: no signal

    Note:
        Sentiment reversals usually lag the exact top or bottom,
        but can still lead by roughly 3-10 trading days.
        Use together with price momentum or volume anomalies.
    """
    signals = pd.DataFrame(index=fg_index.index)
    signals["fg"] = fg_index
    signals["signal"] = 0
    signals["direction"] = ""
    signals["strength"] = 0.0

    # Sustained extreme greed → short signal
    greed_mask = (fg_index > extreme_threshold).rolling(confirmation_days).sum() == confirmation_days
    signals.loc[greed_mask, "signal"] = 1
    signals.loc[greed_mask, "direction"] = "short"
    signals.loc[greed_mask, "strength"] = (fg_index - extreme_threshold).clip(0) / 20

    # Sustained extreme fear → long signal
    fear_mask = (fg_index < fear_threshold).rolling(confirmation_days).sum() == confirmation_days
    signals.loc[fear_mask, "signal"] = -1
    signals.loc[fear_mask, "direction"] = "long"
    signals.loc[fear_mask, "strength"] = (fear_threshold - fg_index).clip(0) / 20

    return signals

def compute_kol_influence_score(author: dict, recent_tweets: list[dict]) -> float:
    """Quantify the market influence of a Twitter KOL.

    Args:
        author: Account metadata including follower count, verification, and age
        recent_tweets: Most recent 20 tweets including engagement metrics

    Returns:
        Influence score in [0, 100]

    Note:
        KOLs with scores above 70 tend to lift 1-hour realized volatility
        of the related asset by roughly 15% after posting.
    """
    # Follower-quality score, using log follower count
    follower_score = min(np.log10(max(author["followers_count"], 1)) / 7, 1.0) * 40

    # Engagement rate = recent average engagement / follower count
    avg_engagement = np.mean([
        t["metrics"]["like_count"] + t["metrics"]["retweet_count"] * 2
        for t in recent_tweets
    ])
    engagement_rate = avg_engagement / max(author["followers_count"], 1)
    engagement_score = min(engagement_rate * 1000, 1.0) * 30

    # Account-age credibility score
    age_score = min(author["account_age_days"] / 1825, 1.0) * 20  # full score at 5 years

    # Verification bonus
    verified_bonus = 10 if author["verified"] else 0

    return follower_score + engagement_score + age_score + verified_bonus

def analyze_earnings_sentiment_shift(
    ticker: str,
    earnings_date: str,
    sentiment_df: pd.DataFrame,
    window_days: int = 5,
) -> dict:
    """Analyze Twitter sentiment changes before and after earnings.

    Args:
        ticker: Stock ticker
        earnings_date: Earnings date in YYYY-MM-DD
        sentiment_df: Daily time series containing date / sentiment_score
        window_days: Observation window on each side of the earnings date

    Returns:
        {
          'pre_sentiment': float,
          'post_sentiment': float,
          'shift': float,
          'signal': str  # 'beat_expected' / 'miss_expected' / 'neutral'
        }
    """
    ed = pd.Timestamp(earnings_date)
    pre = sentiment_df[
        (sentiment_df.index >= ed - pd.Timedelta(days=window_days)) &
        (sentiment_df.index < ed)
    ]["sentiment_score"].mean()
    post = sentiment_df[
        (sentiment_df.index > ed) &
        (sentiment_df.index <= ed + pd.Timedelta(days=window_days))
    ]["sentiment_score"].mean()

    shift = post - pre
    signal = "neutral"
    if shift > 0.2:
        signal = "beat_expected"
    elif shift < -0.2:
        signal = "miss_expected"

    return {"pre_sentiment": pre, "post_sentiment": post, "shift": shift, "signal": signal}

ALPHA_QUALITY_RULES = {
    # Low-quality signals to filter out directly
    "spam_patterns": [
        r"100x guaranteed",
        r"private sale",
        r"limited spots",
        r"DM me",
        r"pump incoming",
    ],
    # High-quality signals that deserve extra weight
    "quality_signals": [
        r"on-chain data",
        r"tokenomics analysis",
        r"team background",
        r"audit report",
        r"TVL growing",
    ],
}

def score_telegram_alpha(message: str) -> dict:
    """Score the quality of alpha in Telegram crypto-channel messages.

    Args:
        message: Raw channel message

    Returns:
        {'quality_score': int[0-10], 'is_spam': bool, 'alpha_type': str}
    """
    import re
    text_lower = message.lower()

    # Spam detection
    for pattern in ALPHA_QUALITY_RULES["spam_patterns"]:
        if re.search(pattern, text_lower):
            return {"quality_score": 0, "is_spam": True, "alpha_type": "spam"}

    # Quality score
    score = 5
    for pattern in ALPHA_QUALITY_RULES["quality_signals"]:
        if re.search(pattern, text_lower):
            score += 1

    alpha_type = "research" if score >= 7 else "signal" if score >= 5 else "noise"
    return {"quality_score": min(score, 10), "is_spam": False, "alpha_type": alpha_type}

def compute_project_health_index(
    discord_stats: dict,
    lookback_days: int = 30,
) -> dict:
    """Build a project-community health index from Discord statistics.

    Args:
        discord_stats: {
            'daily_messages': list[int],
            'daily_active_users': list[int],
            'new_members': list[int],
            'dev_commits': list[int],
        }
        lookback_days: Historical window used as the baseline

    Returns:
        {
            'health_score': float[0-100],
            'trend': 'growing'|'stable'|'declining',
            'flags': list[str]
        }
    """
    msgs = np.array(discord_stats["daily_messages"][-lookback_days:])
    users = np.array(discord_stats["daily_active_users"][-lookback_days:])
    members = np.array(discord_stats["new_members"][-lookback_days:])

    # Component scores
    msg_trend = np.polyfit(range(len(msgs)), msgs, 1)[0]
    user_trend = np.polyfit(range(len(users)), users, 1)[0]

    msg_score = min(50 + msg_trend / max(msgs.mean(), 1) * 500, 100)
    user_score = min(50 + user_trend / max(users.mean(), 1) * 500, 100)
    retention = (users.mean() / max(members[-7:].sum(), 1)) * 20

    health_score = 0.4 * msg_score + 0.4 * user_score + 0.2 * min(retention, 100)

    # Warning flags
    flags = []
    if msgs[-7:].mean() < msgs[-30:].mean() * 0.5:
        flags.append("Message volume has collapsed: 7-day average is below 50% of the 30-day average")
    if users[-3:].mean() < users[-30:].mean() * 0.3:
        flags.append("Active users have plunged: possible project-abandonment warning")

    trend = "growing" if msg_trend > 0 and user_trend > 0 else \
            "declining" if msg_trend < 0 and user_trend < 0 else "stable"

    return {"health_score": health_score, "trend": trend, "flags": flags}

def detect_meme_stock_momentum(
    wsb_posts: list[dict],
    top_n: int = 10,
) -> pd.DataFrame:
    """Detect meme-stock momentum on WSB and identify short-squeeze candidates.

    Args:
        wsb_posts: List of WSB posts collected through PRAW
        top_n: Number of top hot tickers to return

    Returns:
        DataFrame containing ticker / mention_count / avg_score / option_buzz

    Note:
        mention_count > 200 in one day plus sentiment > 0.5 is one early short-squeeze warning condition.
        A full squeeze setup still requires short interest above 20%.
    """
    import re
    from collections import defaultdict

    ticker_stats = defaultdict(lambda: {"count": 0, "scores": [], "option_buzz": 0})

    # Common U.S. ticker regex: 2-5 uppercase letters
    ticker_pattern = re.compile(r"\b([A-Z]{2,5})\b")
    option_keywords = ["calls", "puts", "options", "IV", "yolo", "FDs"]

    for post in wsb_posts:
        text = f"{post['title']} {post['selftext']}"
        tickers_found = ticker_pattern.findall(text)

        # Filter common non-ticker words
        stop_words = {"THE", "FOR", "AND", "BUT", "NOT", "ARE", "YOU", "ALL", "CAN"}
        tickers_found = [t for t in tickers_found if t not in stop_words]

        has_options = any(kw.lower() in text.lower() for kw in option_keywords)

        for ticker in set(tickers_found):
            ticker_stats[ticker]["count"] += 1
            ticker_stats[ticker]["scores"].append(post["score"])
            if has_options:
                ticker_stats[ticker]["option_buzz"] += 1

    rows = []
    for ticker, stats in ticker_stats.items():
        rows.append({
            "ticker": ticker,
            "mention_count": stats["count"],
            "avg_score": np.mean(stats["scores"]),
            "option_buzz": stats["option_buzz"],
        })

    df = pd.DataFrame(rows).sort_values("mention_count", ascending=False)
    return df.head(top_n)

# Reference implementation: agent/src/tools/social_media_tool.py

from dataclasses import dataclass
from enum import Enum

class Platform(str, Enum):
    TWITTER = "twitter"
    TELEGRAM = "telegram"
    DISCORD = "discord"
    REDDIT = "reddit"

@dataclass
class SocialMediaQuery:
    """Social-media query parameters."""
    platform: Platform
    query: str
    limit: int = 100
    start_time: str | None = None
    include_sentiment: bool = True

def collect_social_signals(query: SocialMediaQuery) -> dict:
    """Unified entrypoint for collecting social-media data.

    Args:
        query: Query parameters including platform, keyword, time window, and limit

    Returns:
        Standardized JSON data containing platform / items / metadata

    Raises:
        ValueError: Unsupported platform or invalid parameters
        RuntimeError: API call failed, with retry guidance attached
    """
    collectors = {
        Platform.TWITTER: _collect_twitter,
        Platform.TELEGRAM: _collect_telegram,
        Platform.DISCORD: _collect_discord,
        Platform.REDDIT: _collect_reddit,
    }
    collector = collectors.get(query.platform)
    if not collector:
        raise ValueError(f"Unsupported platform: {query.platform}")

    raw_data = collector(query)

    if query.include_sentiment:
        raw_data = _enrich_with_sentiment(raw_data)

    return raw_data

# Add the following to .env
TWITTER_BEARER_TOKEN=xxx
TELEGRAM_API_ID=xxx
TELEGRAM_API_HASH=xxx
DISCORD_BOT_TOKEN=xxx
REDDIT_CLIENT_ID=xxx
REDDIT_CLIENT_SECRET=xxx

# Sentiment model selection: vader / finbert / llm
SENTIMENT_MODEL=finbert

-- Social-sentiment factor table (DuckDB / SQLite)
CREATE TABLE social_sentiment_factors (
    date        DATE NOT NULL,
    ticker      VARCHAR(20) NOT NULL,
    platform    VARCHAR(20) NOT NULL,
    sentiment   FLOAT,           -- [-1, 1]
    buzz_zscore FLOAT,           -- Buzz Z-score
    fear_greed  FLOAT,           -- [0, 100]
    msg_count   INTEGER,
    author_type VARCHAR(20),     -- institutional / kol / retail
    PRIMARY KEY (date, ticker, platform)
);