84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import List, Dict, Any
|
|
import httpx
|
|
from app.config import settings
|
|
|
|
ALPHA_BASE = "https://www.alphavantage.co/query"
|
|
|
|
|
|
async def fetch_fx_intraday(symbol: str = "XAUUSD", interval: str = "1min") -> List[Dict[str, Any]]:
|
|
from_symbol = symbol[:3].upper()
|
|
to_symbol = symbol[3:].upper()
|
|
params = {
|
|
"function": "FX_INTRADAY",
|
|
"from_symbol": from_symbol,
|
|
"to_symbol": to_symbol,
|
|
"interval": interval,
|
|
"outputsize": "compact",
|
|
"apikey": settings.ALPHA_VANTAGE_API_KEY or "demo",
|
|
}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
r = await client.get(ALPHA_BASE, params=params)
|
|
r.raise_for_status()
|
|
js = r.json()
|
|
key = f"Time Series FX ({interval})"
|
|
series = js.get(key) or {}
|
|
out: List[Dict[str, Any]] = []
|
|
# Alpha returns in reverse chronological; convert to ascending
|
|
for ts, row in reversed(list(series.items())):
|
|
# ts like '2024-11-01 10:05:00'
|
|
# Convert to seconds
|
|
# We avoid datetime parsing heavy ops; split string
|
|
date_part, time_part = ts.split(" ")
|
|
y, m, d = map(int, date_part.split("-"))
|
|
hh, mm, ss = map(int, time_part.split(":"))
|
|
import calendar, datetime as dt
|
|
seconds = int(calendar.timegm(dt.datetime(y, m, d, hh, mm, ss).timetuple()))
|
|
out.append(
|
|
{
|
|
"time": seconds,
|
|
"open": float(row["1. open"]),
|
|
"high": float(row["2. high"]),
|
|
"low": float(row["3. low"]),
|
|
"close": float(row["4. close"]),
|
|
"volume": float(row.get("5. volume", 0.0)),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
async def fetch_fx_daily(symbol: str = "XAUUSD") -> List[Dict[str, Any]]:
|
|
from_symbol = symbol[:3].upper()
|
|
to_symbol = symbol[3:].upper()
|
|
params = {
|
|
"function": "FX_DAILY",
|
|
"from_symbol": from_symbol,
|
|
"to_symbol": to_symbol,
|
|
"outputsize": "compact",
|
|
"apikey": settings.ALPHA_VANTAGE_API_KEY or "demo",
|
|
}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
r = await client.get(ALPHA_BASE, params=params)
|
|
r.raise_for_status()
|
|
js = r.json()
|
|
key = "Time Series FX (Daily)"
|
|
series = js.get(key) or {}
|
|
out: List[Dict[str, Any]] = []
|
|
for ts, row in reversed(list(series.items())):
|
|
# ts like '2024-11-01'
|
|
import calendar, datetime as dt
|
|
y, m, d = map(int, ts.split("-"))
|
|
seconds = int(calendar.timegm(dt.datetime(y, m, d, 0, 0, 0).timetuple()))
|
|
out.append(
|
|
{
|
|
"time": seconds,
|
|
"open": float(row["1. open"]),
|
|
"high": float(row["2. high"]),
|
|
"low": float(row["3. low"]),
|
|
"close": float(row["4. close"]),
|
|
"volume": 0.0,
|
|
}
|
|
)
|
|
return out
|