Spaces:
Sleeping
Sleeping
File size: 2,327 Bytes
e8aaca5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | from typing import Tuple
import pandas as pd
import requests
from fastapi import HTTPException
from app.config import get_settings
from app.indicators import add_technical_indicators
settings = get_settings()
ALPHA_VANTAGE_URL = "https://www.alphavantage.co/query"
def fetch_raw_history(ticker: str) -> pd.DataFrame:
"""
Fetch OHLCV daily data from Alpha Vantage (FREE endpoint).
Replaces yfinance for HF Spaces compatibility.
"""
params = {
"function": "TIME_SERIES_DAILY",
"symbol": ticker,
"apikey": settings.alpha_vantage_api_key,
"outputsize": "compact",
}
try:
response = requests.get(ALPHA_VANTAGE_URL, params=params, timeout=10)
data = response.json()
except Exception:
raise HTTPException(
status_code=503,
detail="Market data provider unavailable",
)
if "Time Series (Daily)" not in data:
raise HTTPException(
status_code=400,
detail="Invalid ticker or API rate limit exceeded",
)
ts = data["Time Series (Daily)"]
df = pd.DataFrame.from_dict(ts, orient="index").astype(float)
df.index = pd.to_datetime(df.index)
df.sort_index(inplace=True)
df.rename(
columns={
"1. open": "Open",
"2. high": "High",
"3. low": "Low",
"4. close": "Close",
"5. volume": "Volume",
},
inplace=True,
)
if df.empty or len(df) < settings.history_window + 60:
raise HTTPException(
status_code=400,
detail="Not enough historical data for this ticker.",
)
return df[["Open", "High", "Low", "Close", "Volume"]]
def get_enriched_history(ticker: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""
Returns:
- Raw OHLCV dataframe
- Technical-indicator-enriched dataframe
"""
df_raw = fetch_raw_history(ticker)
df_tech = add_technical_indicators(df_raw)
return df_raw, df_tech
def last_n_candles(df: pd.DataFrame, n: int) -> list[dict]:
"""
Used by frontend for recent price chart
"""
tail = df.tail(n)
return [
{
"date": idx.strftime("%Y-%m-%d"),
"price": float(row["Close"]),
}
for idx, row in tail.iterrows()
]
|