|
|
@@ -0,0 +1,48 @@
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import dataclasses
|
|
|
+import typing
|
|
|
+
|
|
|
+import cache
|
|
|
+import roi
|
|
|
+
|
|
|
+def main() -> None:
|
|
|
+ sankey: typing.Mapping[str, SankeyMat] = cache.get('https://benten.space/sankey_data.json')
|
|
|
+ raw_prices: list[roi.RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json')
|
|
|
+ prices: dict[str, roi.Price] = {
|
|
|
+ p['MaterialTicker']: roi.Price(p['VWAP7D'], p['AverageTraded7D'], p['VWAP30D']) for p in raw_prices
|
|
|
+ if p['ExchangeCode'] == 'IC1'
|
|
|
+ }
|
|
|
+
|
|
|
+ profits: list[Profit] = []
|
|
|
+ for mat, sankey_data in sankey.items():
|
|
|
+ price = prices[mat]
|
|
|
+ if price.vwap_7d is None:
|
|
|
+ continue
|
|
|
+ assert price.average_traded_7d is not None
|
|
|
+ profits.append(Profit(mat, price.vwap_7d / area_cost(mat, sankey_data), price.average_traded_7d))
|
|
|
+ profits.sort(reverse=True)
|
|
|
+ for profit in profits:
|
|
|
+ print(f'{profit.material:4} {profit.per_area:5.2f} {profit.traded:6.2f}')
|
|
|
+
|
|
|
+def area_cost(mat: str, sankey_data: SankeyMat) -> float:
|
|
|
+ return sum(link['value'] for link in sankey_data['links'] if link['source'] == mat)
|
|
|
+
|
|
|
+class SankeyMat(typing.TypedDict):
|
|
|
+ links: typing.Sequence[SankeyLink]
|
|
|
+
|
|
|
+class SankeyLink(typing.TypedDict):
|
|
|
+ source: str
|
|
|
+ value: float
|
|
|
+
|
|
|
+@dataclasses.dataclass(eq=False, frozen=True, slots=True)
|
|
|
+class Profit:
|
|
|
+ material: str
|
|
|
+ per_area: float
|
|
|
+ traded: float
|
|
|
+
|
|
|
+ def __lt__(self, other: Profit) -> bool:
|
|
|
+ return self.per_area < other.per_area
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ main()
|