alphavantage-api

Alpha Vantage API 集成

金融数据 API，提供股票、外汇、加密货币、技术指标、基本面数据、经济指标，以及 AI 驱动的新闻情绪分析。

快速入门

身份验证

# Environment variable (recommended)
export ALPHAVANTAGE_API_KEY="your_api_key"

# Or in .env file
ALPHAVANTAGE_API_KEY=your_api_key

基本用法（Python）

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", {})

# Example
quote = get_quote("AAPL")
print(f"AAPL: ${quote['05. price']} ({quote['10. change percent']})")

使用 Python 包

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

# Time series data
ts = TimeSeries(key=API_KEY, output_format='pandas')
data, meta = ts.get_daily(symbol='AAPL', outputsize='compact')

# Technical indicators
ti = TechIndicators(key=API_KEY, output_format='pandas')
rsi, meta = ti.get_rsi(symbol='AAPL', interval='daily', time_period=14)

API 函数参考

股票时间序列

函数	描述	免费
TIME_SERIES_INTRADAY	1-60min 间隔	✅
TIME_SERIES_DAILY	每日 OHLCV	✅
TIME_SERIES_DAILY_ADJUSTED	包含拆股／股息调整	⚠️ 高级版
TIME_SERIES_WEEKLY	每周 OHLCV	✅
TIME_SERIES_MONTHLY	每月 OHLCV	✅
GLOBAL_QUOTE	最新报价	✅
SYMBOL_SEARCH	搜索股票代码	✅

基本面数据

函数	描述	免费
OVERVIEW	公司概况	✅
INCOME_STATEMENT	损益表	✅
BALANCE_SHEET	资产负债表	✅
CASH_FLOW	现金流量表	✅
EARNINGS	盈利历史	✅
EARNINGS_CALENDAR	即将公布的盈利	✅
IPO_CALENDAR	即将进行的 IPO	✅

外汇

函数	描述	免费
CURRENCY_EXCHANGE_RATE	实时汇率	✅
FX_INTRADAY	盘中外汇数据	✅
FX_DAILY	每日外汇数据	✅
FX_WEEKLY	每周外汇数据	✅
FX_MONTHLY	每月外汇数据	✅

加密货币

函数	描述	免费
CURRENCY_EXCHANGE_RATE	加密货币汇率	✅
DIGITAL_CURRENCY_DAILY	每日加密货币数据	✅
DIGITAL_CURRENCY_WEEKLY	每周加密货币数据	✅
DIGITAL_CURRENCY_MONTHLY	每月加密货币数据	✅

技术指标（50+）

类别	指标
趋势	SMA、EMA、WMA、DEMA、TEMA、KAMA、MAMA、T3、TRIMA
动量	RSI、MACD、STOCH、WILLR、ADX、CCI、MFI、ROC、AROON、MOM
波动率	BBANDS、ATR、NATR、TRANGE
成交量	OBV、AD、ADOSC
希尔伯特	HT_TRENDLINE、HT_SINE、HT_PHASOR 等

经济指标

函数	描述	免费
REAL_GDP	美国 GDP	✅
CPI	消费者价格指数	✅
INFLATION	通胀率	✅
UNEMPLOYMENT	失业率	✅
FEDERAL_FUNDS_RATE	联邦基金利率	✅
TREASURY_YIELD	国债收益率	✅

Alpha Intelligence

函数	描述	免费
NEWS_SENTIMENT	AI 情绪分析	✅
TOP_GAINERS_LOSERS	市场涨跌幅榜	✅
INSIDER_TRANSACTIONS	内部人交易	⚠️ 高级版
ANALYTICS_FIXED_WINDOW	分析	⚠️ 高级版

速率限制

套餐	每日	每分钟	价格
免费	25	5	$0
高级版	不限	75-1,200	$49.99-$249.99/mo

**重要：**速率限制按 IP 地址计算，而非按密钥计算。

常见任务

任务：获取每日股票数据

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)", {})

# Example
data = get_daily_data("AAPL")
latest = list(data.items())[0]
print(f"{latest[0]}: Close ${latest[1]['4. close']}")

任务：获取技术指标

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", {})

# Example
rsi = get_rsi("AAPL")
latest_rsi = list(rsi.values())[0]["RSI"]
print(f"AAPL RSI(14): {latest_rsi}")

任务：获取公司概况

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", {})

# Example
rate = get_forex_rate("USD", "EUR")
print(f"USD/EUR: {rate['5. Exchange Rate']}")

任务：获取加密货币价格

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")
    }

# Example
btc = get_crypto_price("BTC")
print(f"BTC: ${float(btc['price']):,.2f}")

任务：获取新闻情绪

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", [])

# Example
news = get_news_sentiment(tickers="AAPL")
for article in news[:3]:
    sentiment = article.get("overall_sentiment_label", "N/A")
    print(f"{article['title'][:50]}... [{sentiment}]")

任务：获取经济指标

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()

# Examples
gdp = get_economic_indicator("REAL_GDP")
cpi = get_economic_indicator("CPI")
unemployment = get_economic_indicator("UNEMPLOYMENT")
fed_rate = get_economic_indicator("FEDERAL_FUNDS_RATE")

任务：获取盈利日历

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)

# Example
earnings = get_earnings_calendar()
for e in earnings[:5]:
    print(f"{e['symbol']}: {e['reportDate']}")

错误处理

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 {}

免费版与高级版功能

免费版包含

• 每天 25 次请求
• 每分钟 5 次请求
• 历史时间序列（20+ 年）
• 50+ 项技术指标
• 基本面数据
• 外汇和加密货币
• 经济指标
• 新闻情绪

需要高级版

• 每日请求次数不限
• 调整后的时间序列
• 美国市场实时数据
• 延迟 15 分钟的数据
• 内部人交易
• 高级分析
• 优先支持

最佳实践

1. 缓存响应 - 数据不会频繁变化
2. 使用紧凑的 outputsize - 除非需要完整历史记录
3. 合理批量请求 - 每日 25 次的限制非常严格
4. 处理速率限制 - 检查响应中的 "Note" 键
5. 使用 pandas 输出 - 配合 alpha_vantage 包使用
6. 存储历史数据 - 避免重复获取相同数据

安装

# Official Python wrapper
pip install alpha_vantage pandas

# For async support
pip install aiohttp

使用 Pandas

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

# Initialize with pandas output
ts = TimeSeries(key=API_KEY, output_format='pandas')
ti = TechIndicators(key=API_KEY, output_format='pandas')

# Get daily data
data, meta = ts.get_daily(symbol='AAPL', outputsize='compact')

# Get indicators
sma, _ = ti.get_sma(symbol='AAPL', interval='daily', time_period=20)
rsi, _ = ti.get_rsi(symbol='AAPL', interval='daily', time_period=14)

# Combine
analysis = data.join([sma, rsi])

相关技能

• finnhub-api - 实时报价和新闻
• twelvedata-api - 更多指标，更宽松的速率限制
• fmp-api - 专注于基本面分析

参考资料

• 官方文档
• Python 包
• API 支持