AI Agents · 课时

市场数据 API 集成

Q: 「市场数据 API 集成」这节课中我会学到什么？

使用 Alpha Vantage、Yahoo Finance 和 Polygon.io 获取实时与历史数据 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

使用 Alpha Vantage、Yahoo Finance 和 Polygon.io 获取实时与历史数据

市场数据 API 集成 是 CoddyKit 上的免费 AI Agents 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

金融智能体中的市场数据

金融分析智能体需要访问实时和历史市场数据。常见的数据类型包括 OHLCV（开盘价/最高价/最低价/收盘价/成交量）、财报日期、基本面比率和期权链。

有多种应用程序接口可以提供这些数据：yfinance（免费，数据来自 Yahoo Finance）、Polygon.io（付费，可靠）和Alpha Vantage（提供免费层级）。

yfinance：历史价格数据

yfinance封装了 Yahoo Finance 应用程序接口，是最快的入门方式。Ticker.history()会返回包含 OHLCV 数据的 pandas DataFrame。

import yfinance as yf

ticker = yf.Ticker('AAPL')

# 1 year of daily data
history = ticker.history(period='1y')
print(history.tail(3))
#                  Open        High         Low       Close    Volume
# Date
# 2026-05-27  189.1500  190.3200  188.9200  190.0500  55234000
# 2026-05-28  190.4200  191.5600  189.7800  191.1200  62345000
print(f'Rows: {len(history)}')

yfinance：自定义日期范围

使用 start 和 end 参数指定自定义日期范围，或使用 interval 获取日内数据（1m、5m、1h）。请注意，根据时间间隔的不同，日内数据通常仅限最近 60 天。

import yfinance as yf

ticker = yf.Ticker('MSFT')

# Custom date range
history = ticker.history(start='2024-01-01', end='2024-12-31')
print(f'Trading days in 2024: {len(history)}')

# Batch multiple tickers
tickers = yf.download(['AAPL', 'MSFT', 'GOOGL'],
                       start='2025-01-01', end='2025-12-31')
print(tickers['Close'].head())

Polygon.io REST 接口

Polygon.io 提供专业级市场数据，其速率限制和运行可靠性都高于免费来源。生产环境中的金融智能体应使用它。

import requests

POLYGON_KEY = 'YOUR_POLYGON_API_KEY'

def get_polygon_daily(ticker: str, from_date: str, to_date: str) -> list[dict]:
    url = f'https://api.polygon.io/v2/aggs/ticker/{ticker}/range/1/day/{from_date}/{to_date}'
    resp = requests.get(url, params={'apiKey': POLYGON_KEY, 'limit': 365})
    resp.raise_for_status()
    data = resp.json()
    return data.get('results', [])

bars = get_polygon_daily('AAPL', '2025-01-01', '2025-12-31')
print(f'Bars returned: {len(bars)}')
if bars:
    print('Latest close:', bars[-1]['c'])

Alpha Vantage TIME_SERIES_DAILY

Alpha Vantage 提供免费层级（每天 25 次调用）和付费层级。TIME_SERIES_DAILY端点返回 OHLCV 数据，并可选择返回调整后价格。

import requests

ALPHA_KEY = 'YOUR_ALPHA_VANTAGE_KEY'

def get_alpha_vantage_daily(symbol: str, outputsize: str = 'compact') -> dict:
    # outputsize: 'compact' (100 days) or 'full' (20+ years)
    resp = requests.get('https://www.alphavantage.co/query', params={
        'function':   'TIME_SERIES_DAILY_ADJUSTED',
        'symbol':     symbol,
        'outputsize': outputsize,
        'apikey':     ALPHA_KEY
    })
    resp.raise_for_status()
    data = resp.json()
    series = data.get('Time Series (Daily)', {})
    return series   # dict: {date_str: {open, high, low, close, volume, ...}}

速率限制与重试逻辑

所有市场数据应用程序接口都会实施速率限制。超过限制后，您会收到 429（请求过多）错误。请始终实现指数退避，并遵守速率限制。

import time, requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def make_rate_limited_session(calls_per_minute: int = 5) -> requests.Session:
    session = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=2,         # 1s, 2s, 4s waits
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('https://', adapter)
    session._calls_per_min = calls_per_minute
    session._min_interval  = 60.0 / calls_per_minute
    session._last_call     = 0.0
    return session

def rate_limited_get(session, url, **kwargs):
    elapsed = time.time() - session._last_call
    if elapsed < session._min_interval:
        time.sleep(session._min_interval - elapsed)
    session._last_call = time.time()
    return session.get(url, **kwargs)

缓存市场数据

历史数据不会改变——昨天的收盘价不会变化。请对其进行缓存，避免重复调用应用程序接口。可以使用以（股票代码、date_range）为键的简单磁盘缓存。

import hashlib, json, os

CACHE_DIR = '/tmp/market_cache'
os.makedirs(CACHE_DIR, exist_ok=True)

def cache_key(ticker: str, start: str, end: str) -> str:
    return hashlib.md5(f'{ticker}_{start}_{end}'.encode()).hexdigest()

def get_cached(ticker: str, start: str, end: str):
    key = cache_key(ticker, start, end)
    path = os.path.join(CACHE_DIR, f'{key}.json')
    if os.path.exists(path):
        with open(path) as f:
            return json.load(f)
    return None

def set_cached(ticker: str, start: str, end: str, data):
    key = cache_key(ticker, start, end)
    path = os.path.join(CACHE_DIR, f'{key}.json')
    with open(path, 'w') as f:
        json.dump(data, f)

if __name__ == '__main__':
    set_cached('AAPL', '2024-01-01', '2024-01-31', {'close': [150, 151, 149]})
    cached = get_cached('AAPL', '2024-01-01', '2024-01-31')
    print('Cached data for AAPL:', cached)
    print('Cache miss for MSFT:', get_cached('MSFT', '2024-01-01', '2024-01-31'))

处理交易时间与市场数据缺口

市场在周末和节假日休市。跨越非交易日的日期范围会在数据中形成缺口。请始终检测并处理这些缺口，不要假设每日数据是连续的。

import pandas as pd
from pandas.tseries.offsets import BDay

def detect_gaps(history: pd.DataFrame) -> list[str]:
    if history.empty:
        return []
    date_range = pd.date_range(
        start=history.index.min(),
        end=history.index.max(),
        freq=BDay()  # Business days only
    )
    missing = date_range.difference(history.index)
    return [str(d.date()) for d in missing]

import yfinance as yf
history = yf.Ticker('AAPL').history(start='2024-12-23', end='2025-01-07')
gaps = detect_gaps(history)
print('Missing business days (holidays):', gaps)

为智能体构建市场数据工具

将市场数据获取器注册为智能体工具。LLM 可以调用它来获取数据，然后使用另一个工具（分析工具）处理结果。

import yfinance as yf, json

def get_market_data_tool(ticker: str, period: str = '1y') -> str:
    cached = get_cached(ticker, period, 'yfinance')
    if cached:
        return json.dumps(cached)
    try:
        hist = yf.Ticker(ticker).history(period=period)
        if hist.empty:
            return json.dumps({'error': f'No data for {ticker}'})
        result = {
            'ticker':     ticker,
            'period':     period,
            'start':      str(hist.index.min().date()),
            'end':        str(hist.index.max().date()),
            'latest_close': float(hist['Close'].iloc[-1]),
            'pct_change_ytd': float((hist['Close'].iloc[-1] / hist['Close'].iloc[0] - 1) * 100),
            'rows':       len(hist)
        }
        set_cached(ticker, period, 'yfinance', result)
        return json.dumps(result)
    except Exception as e:
        return json.dumps({'error': str(e)})

来自 yfinance 的基本面数据

除了价格数据之外，yfinance 还提供基本面数据，例如市盈率、每股收益和市值等。基本面数据可以让智能体的分析超越单纯的价格走势。

import yfinance as yf

ticker = yf.Ticker('AAPL')
info  = ticker.info

print('Market Cap:      ', info.get('marketCap'))
print('P/E Ratio:       ', info.get('trailingPE'))
print('EPS:             ', info.get('trailingEps'))
print('52-Week High:    ', info.get('fiftyTwoWeekHigh'))
print('52-Week Low:     ', info.get('fiftyTwoWeekLow'))
print('Dividend Yield:  ', info.get('dividendYield'))
print('Analyst Target:  ', info.get('targetMeanPrice'))

错误处理：已退市和无效的股票代码

并非所有股票代码都有效或仍在交易。已退市公司的历史记录为空。继续分析前，请始终验证返回结果。

import yfinance as yf

def safe_fetch(ticker_symbol: str, period: str = '1y') -> dict:
    try:
        ticker = yf.Ticker(ticker_symbol)
        history = ticker.history(period=period)

        if history.empty:
            return {
                'error': f'No data for {ticker_symbol}. May be delisted or invalid.',
                'ticker': ticker_symbol
            }
        return {
            'ticker':       ticker_symbol,
            'latest_close': float(history['Close'].iloc[-1]),
            'data_start':   str(history.index.min().date()),
            'bars':         len(history)
        }
    except Exception as e:
        return {'error': str(e), 'ticker': ticker_symbol}

print(safe_fetch('AAPL'))    # valid
print(safe_fetch('INVALID')) # returns error dict

Alpha Vantage TIME_SERIES_DAILY 端点中的“compact”输出大小会返回什么

理解应用程序接口的输出大小参数，对于管理金融智能体中的带宽、延迟和应用程序接口配额非常重要。

市场数据应用程序接口集成回顾

使用 yfinance 进行快速原型开发（免费），使用 Polygon.io 进行生产部署（可靠、付费），使用 Alpha Vantage 作为免费层级选项。请始终实现速率限制，为历史数据使用磁盘缓存，并针对非交易日进行缺口检测。

将数据获取功能作为智能体工具公开，使 LLM 能够在需要时请求数据。

常见问题解答

「市场数据 API 集成」课时是免费的吗？

是的 — 「市场数据 API 集成」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「市场数据 API 集成」这节课中我会学到什么？

使用 Alpha Vantage、Yahoo Finance 和 Polygon.io 获取实时与历史数据你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「市场数据 API 集成」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents