Integration von Marktdaten-APIs

Marktdaten in Finanzagenten

yfinance: Historische Kursdaten

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: Benutzerdefinierte Zeiträume

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-API

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

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, ...}}

Ratenlimits und Wiederholungslogik

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)

Marktdaten zwischenspeichern

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

Handelszeiten und Marktlücken berücksichtigen

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)

Ein Marktdaten-Tool für Agenten erstellen

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

Fundamentaldaten aus 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'))

Fehlerbehandlung: Dekotierte und ungültige Ticker

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

Welche Daten gibt die Ausgabegröße „compact“ im Alpha-Vantage-Endpunkt TIME_SERIES_DAILY zurück?

Zusammenfassung zur Integration von Marktdaten-APIs