/
conf.py
143 lines (123 loc) · 4.61 KB
/
conf.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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
from __future__ import annotations
import datetime
import os
from collections.abc import Iterable, Iterator, Mapping
from dataclasses import replace
import yaml
from common import Action, Symbol, tzinfo
_symbols: dict[str, Symbol] = {}
_actions: dict[str, Action] = {}
def symbols() -> dict[str, Symbol]:
return _symbols
def actions() -> dict[str, Action]:
return _actions
class Error(Exception):
def __init__(self, message: str):
super().__init__("config: " + message)
def load(filename: str | os.PathLike[str]) -> Exception | None:
try:
with open(filename, "r") as f:
co = yaml.safe_load(f)
except OSError as err:
return err
except yaml.YAMLError as err:
return err
co = co if co is not None else {}
symbols: dict[str, Symbol] = {}
actions: dict[str, Action] = {}
try:
for symbol in _walk_symbols(co.get("symbols")):
if symbol.name.lower() in symbols:
raise Error(f"duplicate symbol: {symbol.name}")
symbols[symbol.name.lower()] = symbol
for action in _walk_actions(co.get("actions")):
if action.name in actions:
raise Error(f"duplicate action: {action.name}")
actions[action.name] = action
except Error as err:
return err
global _symbols, _actions
_symbols, _actions = symbols, actions
return None
def _walk_symbols(
node: Iterable[Mapping[str, object]] | None,
symbol: Symbol | None = None,
) -> Iterator[Symbol]:
if node is None:
return
if symbol is None:
symbol = Symbol("")
for item in node:
name, symbols = item.get("name"), item.get("symbols")
if name is not None and symbols is not None:
raise Error(f"both name and more symbols in same node with name: {name}")
for k, v in item.items():
match k:
case "name" | "symbols":
pass
case "market":
if type(v) != str:
raise TypeError
symbol.market = v
case "time":
if type(v) != str:
raise TypeError
if tzinfo(v) is None:
raise Error(f"unknown time zone: {v}")
symbol.time = v
case "start":
if type(v) == int:
symbol.start = datetime.datetime(v, 1, 1)
elif type(v) == str:
symbol.start = datetime.datetime.fromisoformat(v)
elif type(v) == datetime.date:
symbol.start = datetime.datetime.combine(v, datetime.time.min)
elif type(v) == datetime.datetime:
symbol.start = v
else:
raise TypeError
case _:
raise Error(f"unexpected key: {k}")
if symbol.start is not None and symbol.start.tzinfo is None:
if symbol.time is not None:
symbol.start = symbol.start.replace(tzinfo=tzinfo(symbol.time))
else:
symbol.start = symbol.start.replace(tzinfo=datetime.timezone.utc)
if name is not None:
if type(name) != str:
raise TypeError
yield replace(symbol, name=name)
elif symbols is not None:
if not isinstance(symbols, Iterable):
raise TypeError
yield from _walk_symbols(symbols, replace(symbol))
def _walk_actions(
node: Iterable[Mapping[str, object]] | None,
action: Action | None = None,
) -> Iterator[Action]:
if node is None:
return
if action is None:
action = Action("")
for item in node:
name, actions = item.get("name"), item.get("actions")
if name is not None and actions is not None:
raise Error(f"both name and more actions in same node with name: {name}")
for k, v in item.items():
match k:
case "name" | "actions":
pass
case "using":
if type(v) != str:
raise TypeError
action.using = v
case _:
raise Error(f"unexpected key: {k}")
if name is not None:
if type(name) != str:
raise TypeError
yield replace(action, name=name)
elif actions is not None:
if not isinstance(actions, Iterable):
raise TypeError
yield from _walk_actions(actions, replace(action))