Alternative Data Integrator

Alternative Data Integrator

import pandas as pd
import numpy as np
from datetime import datetime

class AlternativeDataSources:
    """
    Framework for integrating alternative data. In Claude context, use web_search
    to fetch data, then process through these analytical pipelines.
    """

    # Web search queries for alt data
    SEARCH_QUERIES = {
        "google_trends": "Google Trends {keyword} interest over time",
        "baltic_dry": "Baltic Dry Index today shipping",
        "economic_surprise": "Citigroup Economic Surprise Index",
        "credit_spreads": "US high yield credit spread OAS today",
        "copper_gold_ratio": "copper gold ratio economic indicator",
        "shipping_rates": "container shipping rates index",
        "job_postings": "Indeed job postings trend {country}",
        "restaurant_bookings": "OpenTable restaurant bookings trend",
        "electricity_consumption": "electricity consumption {country} trend",
    }

    @staticmethod
    def google_trends_signal(trend_data: pd.Series, asset: str) -> dict:
        """Process Google Trends data into trading signal.
        Rising search interest often leads price moves by 1-4 weeks."""
        if len(trend_data)  2 else "NORMAL",
            "note": "Google Trends leads retail flows by 1-4 weeks. Contrarian at extremes.",
        }

    @staticmethod
    def economic_nowcast(indicators: dict) -> dict:
        """Combine real-time indicators for economic activity nowcast."""
        scores = {
            "baltic_dry_change": indicators.get("baltic_dry_mom", 0) * 0.15,
            "credit_spread_change": -indicators.get("credit_spread_change", 0) * 0.20,
            "copper_gold_ratio_change": indicators.get("copper_gold_mom", 0) * 0.20,
            "job_postings_change": indicators.get("job_postings_mom", 0) * 0.15,
            "electricity_change": indicators.get("electricity_mom", 0) * 0.10,
            "shipping_rates_change": indicators.get("shipping_mom", 0) * 0.10,
            "consumer_traffic_change": indicators.get("consumer_traffic_mom", 0) * 0.10,
        }
        composite = sum(scores.values())
        return {
            "nowcast_score": round(composite, 4),
            "components": scores,
            "regime": "EXPANSION" if composite > 0.02 else "CONTRACTION" if composite  0.02
                            else "Risk-off currencies favored (JPY, CHF, USD)" if composite  dict:
        """Map search volume patterns to market sentiment."""
        fear_keywords = ["recession", "market crash", "financial crisis", "bank run"]
        greed_keywords = ["bull market", "stock tips", "get rich", "crypto moon"]
        fear_score = sum(keywords.get(k, 0) for k in fear_keywords)
        greed_score = sum(keywords.get(k, 0) for k in greed_keywords)
        net = greed_score - fear_score
        return {
            "fear_index": fear_score,
            "greed_index": greed_score,
            "net_sentiment": round(net, 2),
            "interpretation": "FEAR dominant — contrarian buy signal" if net  50
                            else "BALANCED",
        }

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: mahmoud20138
• Source: mahmoud20138/Tradecraft
• License: MIT

Install and usage instructions live in the source repository linked above.