/
markets.py
104 lines (82 loc) · 3.71 KB
/
markets.py
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
92
93
94
95
96
97
98
99
100
101
102
103
104
from typing import Dict
import timely_beliefs as tb
from timely_beliefs.sensors.func_store import knowledge_horizons
from sqlalchemy.orm import Query
from flexmeasures.data.config import db
from flexmeasures.data.models.time_series import TimedValue
from flexmeasures.utils.flexmeasures_inflection import humanize
class MarketType(db.Model):
"""Describing market types for our purposes.
TODO: Add useful attributes like frequency (e.g. 1H) and the meaning of units (e.g. Mwh).
"""
name = db.Column(db.String(80), primary_key=True)
display_name = db.Column(db.String(80), default="", unique=True)
daily_seasonality = db.Column(db.Boolean(), nullable=False, default=False)
weekly_seasonality = db.Column(db.Boolean(), nullable=False, default=False)
yearly_seasonality = db.Column(db.Boolean(), nullable=False, default=False)
def __init__(self, **kwargs):
super(MarketType, self).__init__(**kwargs)
self.name = self.name.replace(" ", "_").lower()
if "display_name" not in kwargs:
self.display_name = humanize(self.name)
@property
def preconditions(self) -> Dict[str, bool]:
"""Assumptions about the time series data set, such as normality and stationarity
For now, this is usable input for Prophet (see init), but it might evolve or go away."""
return dict(
daily_seasonality=self.daily_seasonality,
weekly_seasonality=self.weekly_seasonality,
yearly_seasonality=self.yearly_seasonality,
)
def __repr__(self):
return "<MarketType %r>" % self.name
class Market(db.Model, tb.SensorDBMixin):
"""Each market is a pricing service."""
name = db.Column(db.String(80), unique=True)
display_name = db.Column(db.String(80), default="", unique=True)
market_type_name = db.Column(
db.String(80), db.ForeignKey("market_type.name"), nullable=False
)
def __init__(self, **kwargs):
# Set default knowledge horizon function for an economic sensor
if "knowledge_horizon_fnc" not in kwargs:
kwargs["knowledge_horizon_fnc"] = knowledge_horizons.ex_ante.__name__
if "knowledge_horizon_par" not in kwargs:
kwargs["knowledge_horizon_par"] = {
knowledge_horizons.ex_ante.__code__.co_varnames[1]: "PT0H"
}
super(Market, self).__init__(**kwargs)
self.name = self.name.replace(" ", "_").lower()
if "display_name" not in kwargs:
self.display_name = humanize(self.name)
@property
def price_unit(self) -> str:
"""Return the 'unit' property of the generic asset, just with a more insightful name."""
return self.unit
market_type = db.relationship(
"MarketType", backref=db.backref("markets", lazy=True)
)
def __repr__(self):
return "<Market %s:%r (%r) res.: %s>" % (
self.id,
self.name,
self.market_type_name,
self.event_resolution,
)
def to_dict(self) -> Dict[str, str]:
return dict(name=self.name, market_type=self.market_type.name)
class Price(TimedValue, db.Model):
"""
All prices are stored in one slim table.
TODO: datetime objects take up most of the space (12 bytes each)). One way out is to normalise them out to a table.
"""
market_id = db.Column(
db.Integer(), db.ForeignKey("market.id"), primary_key=True, index=True
)
market = db.relationship("Market", backref=db.backref("prices", lazy=True))
@classmethod
def make_query(cls, **kwargs) -> Query:
"""Construct the database query."""
return super().make_query(asset_class=Market, **kwargs)
def __init__(self, **kwargs):
super(Price, self).__init__(**kwargs)