undefined

undefined

import requests
import os

API_KEY = os.getenv("ALPHAVANTAGE_API_KEY")
BASE_URL = "https://www.alphavantage.co/query"

def get_quote(symbol: str) -> dict:
    """Get real-time quote for a symbol."""
    response = requests.get(BASE_URL, params={
        "function": "GLOBAL_QUOTE",
        "symbol": symbol,
        "apikey": API_KEY
    })
    return response.json().get("Global Quote", {})

undefined

from alpha_vantage.timeseries import TimeSeries
from alpha_vantage.techindicators import TechIndicators

undefined

def get_daily_data(symbol: str, full: bool = False) -> dict:
    """Get daily OHLCV data."""
    response = requests.get(BASE_URL, params={
        "function": "TIME_SERIES_DAILY",
        "symbol": symbol,
        "outputsize": "full" if full else "compact",
        "apikey": API_KEY
    })
    return response.json().get("Time Series (Daily)", {})

undefined

def get_rsi(symbol: str, period: int = 14) -> dict:
    """Get RSI indicator values."""
    response = requests.get(BASE_URL, params={
        "function": "RSI",
        "symbol": symbol,
        "interval": "daily",
        "time_period": period,
        "series_type": "close",
        "apikey": API_KEY
    })
    return response.json().get("Technical Analysis: RSI", {})

undefined

def get_company_overview(symbol: str) -> dict:
    """Get comprehensive company information."""
    response = requests.get(BASE_URL, params={
        "function": "OVERVIEW",
        "symbol": symbol,
        "apikey": API_KEY
    })
    data = response.json()

    return {
        "name": data.get("Name"),
        "description": data.get("Description"),
        "sector": data.get("Sector"),
        "industry": data.get("Industry"),
        "market_cap": data.get("MarketCapitalization"),
        "pe_ratio": data.get("PERatio"),
        "dividend_yield": data.get("DividendYield"),
        "eps": data.get("EPS"),
        "52_week_high": data.get("52WeekHigh"),
        "52_week_low": data.get("52WeekLow"),
        "beta": data.get("Beta")
    }

def get_forex_rate(from_currency: str, to_currency: str) -> dict:
    """Get currency exchange rate."""
    response = requests.get(BASE_URL, params={
        "function": "CURRENCY_EXCHANGE_RATE",
        "from_currency": from_currency,
        "to_currency": to_currency,
        "apikey": API_KEY
    })
    return response.json().get("Realtime Currency Exchange Rate", {})

undefined

def get_crypto_price(symbol: str, market: str = "USD") -> dict:
    """Get cryptocurrency price."""
    response = requests.get(BASE_URL, params={
        "function": "CURRENCY_EXCHANGE_RATE",
        "from_currency": symbol,
        "to_currency": market,
        "apikey": API_KEY
    })
    data = response.json().get("Realtime Currency Exchange Rate", {})
    return {
        "symbol": symbol,
        "price": data.get("5. Exchange Rate"),
        "last_updated": data.get("6. Last Refreshed")
    }

undefined

def get_news_sentiment(tickers: str = None, topics: str = None) -> list:
    """Get AI-powered news sentiment analysis."""
    params = {
        "function": "NEWS_SENTIMENT",
        "apikey": API_KEY
    }
    if tickers:
        params["tickers"] = tickers
    if topics:
        params["topics"] = topics

    response = requests.get(BASE_URL, params=params)
    return response.json().get("feed", [])

undefined

def get_economic_indicator(indicator: str) -> dict:
    """Get US economic indicator data."""
    response = requests.get(BASE_URL, params={
        "function": indicator,
        "apikey": API_KEY
    })
    return response.json()

undefined

def get_earnings_calendar(horizon: str = "3month") -> list:
    """Get upcoming earnings releases."""
    import csv
    from io import StringIO

    response = requests.get(BASE_URL, params={
        "function": "EARNINGS_CALENDAR",
        "horizon": horizon,  # 3month, 6month, 12month
        "apikey": API_KEY
    })

    # Returns CSV format
    reader = csv.DictReader(StringIO(response.text))
    return list(reader)

undefined

def safe_api_call(params: dict) -> dict:
    """Make API call with error handling."""
    params["apikey"] = API_KEY

    try:
        response = requests.get(BASE_URL, params=params)
        data = response.json()

        # Check for rate limit
        if "Note" in data:
            print(f"Rate limit: {data['Note']}")
            return {}

        # Check for error message
        if "Error Message" in data:
            print(f"API Error: {data['Error Message']}")
            return {}

        # Check for information message (often rate limit)
        if "Information" in data:
            print(f"Info: {data['Information']}")
            return {}

        return data

    except Exception as e:
        print(f"Request error: {e}")
        return {}

undefined

undefined

from alpha_vantage.timeseries import TimeSeries
from alpha_vantage.techindicators import TechIndicators
import pandas as pd

undefined