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
|
from fastapi import APIRouter, Query, HTTPException
from models.market import MarketData
from gridstatus import ISONE, CAISO, Ercot, MISO, NYISO, PJM
from datetime import datetime, timedelta
from typing import List, Dict, Type
router = APIRouter()
MARKET_CLASSES: Dict[str, Type] = {
"ISONE": ISONE,
# "CAISO": CAISO,
# "ERCOT": Ercot,
"MISO": MISO,
"NYISO": NYISO,
# "PJM": PJM,
}
# In-memory cache
_cached_day_ahead: Dict[str, List[MarketData]] = {}
_cached_day_ahead_timestamp: Dict[str, datetime] = {}
_cached_real_time: Dict[str, List[MarketData]] = {}
_cached_real_time_timestamp: Dict[str, datetime] = {}
def get_iso_instance(market: str):
market = market.upper()
if market not in MARKET_CLASSES:
raise HTTPException(status_code=400, detail=f"Unsupported market '{market}'. Supported: {list(MARKET_CLASSES.keys())}")
return MARKET_CLASSES[market]()
@router.get("/day-ahead", response_model=List[MarketData])
def get_day_ahead_data(market: str = Query("ISONE")):
now = datetime.utcnow()
market = market.upper()
if (market not in _cached_day_ahead_timestamp or
now - _cached_day_ahead_timestamp[market] > timedelta(hours=1)):
iso = get_iso_instance(market)
df = iso.get_lmp(date=datetime.now().date(), market="DAY_AHEAD_HOURLY", locations="ALL")
grouped = df.groupby("Interval Start")[["LMP", "Energy", "Congestion", "Loss"]].mean().reset_index()
_cached_day_ahead[market] = [
MarketData(
timestamp=row["Interval Start"],
lmp=row["LMP"],
energy=row["Energy"],
congestion=row["Congestion"],
loss=row["Loss"],
)
for _, row in grouped.iterrows()
]
_cached_day_ahead_timestamp[market] = now
return _cached_day_ahead[market]
@router.get("/real-time", response_model=List[MarketData])
def get_real_time_data(market: str = Query("ISONE")):
now = datetime.utcnow()
market = market.upper()
if (market not in _cached_real_time_timestamp or
now - _cached_real_time_timestamp[market] > timedelta(minutes=5)):
iso = get_iso_instance(market)
df = iso.get_lmp(date="today", market="REAL_TIME_5_MIN", locations="ALL")
grouped = df.groupby("Interval Start")[["LMP", "Energy", "Congestion", "Loss"]].mean().reset_index()
_cached_real_time[market] = [
MarketData(
timestamp=row["Interval Start"],
lmp=row["LMP"],
energy=row["Energy"],
congestion=row["Congestion"],
loss=row["Loss"],
)
for _, row in grouped.iterrows()
]
_cached_real_time_timestamp[market] = now
return _cached_real_time[market]
|
