|
@@ -18,7 +18,6 @@ def main() -> None:
|
|
|
|
|
|
|
|
def calc_for_cx(cx: str, recipes: typing.Collection[Recipe], buildings: typing.Mapping[str, Building],
|
|
def calc_for_cx(cx: str, recipes: typing.Collection[Recipe], buildings: typing.Mapping[str, Building],
|
|
|
materials: typing.Mapping[str, Material], raw_prices: typing.Collection[RawPrice]) -> typing.Sequence[Profit]:
|
|
materials: typing.Mapping[str, Material], raw_prices: typing.Collection[RawPrice]) -> typing.Sequence[Profit]:
|
|
|
- # EXTREME DETAIL: We extract 'Bid' and 'Ask' directly from the raw_prices API payload.
|
|
|
|
|
prices: dict[str, Price] = {
|
|
prices: dict[str, Price] = {
|
|
|
p['MaterialTicker']: Price(p['VWAP7D'], p['AverageTraded7D'], p['VWAP30D'], p['Bid'], p['Ask']) for p in raw_prices # pyright: ignore[reportArgumentType]
|
|
p['MaterialTicker']: Price(p['VWAP7D'], p['AverageTraded7D'], p['VWAP30D'], p['Bid'], p['Ask']) for p in raw_prices # pyright: ignore[reportArgumentType]
|
|
|
if p['ExchangeCode'] == cx
|
|
if p['ExchangeCode'] == cx
|
|
@@ -45,8 +44,6 @@ def calc_for_cx(cx: str, recipes: typing.Collection[Recipe], buildings: typing.M
|
|
|
return profits
|
|
return profits
|
|
|
|
|
|
|
|
def get_metrics(amount: float, price: Price) -> dict[str, float]:
|
|
def get_metrics(amount: float, price: Price) -> dict[str, float]:
|
|
|
- # EXTREME DETAIL: Helper function to generate a 3-part dictionary (VWAP, Bid, Ask) for any item quantity.
|
|
|
|
|
- # If Bid or Ask data is absent from the exchange, we gracefully fall back to the VWAP to prevent math crashes.
|
|
|
|
|
v = price.vwap_7d or price.vwap_30d or 0.0
|
|
v = price.vwap_7d or price.vwap_30d or 0.0
|
|
|
b = price.bid if price.bid is not None else v
|
|
b = price.bid if price.bid is not None else v
|
|
|
a = price.ask if price.ask is not None else v
|
|
a = price.ask if price.ask is not None else v
|
|
@@ -73,7 +70,6 @@ def calc_profit(recipe: Recipe, buildings: typing.Mapping[str, Building], hab_ar
|
|
|
return # skip recipes with thinly traded outputs
|
|
return # skip recipes with thinly traded outputs
|
|
|
output_prices[output['material_ticker']] = typing.cast(PriceNonNull, price)
|
|
output_prices[output['material_ticker']] = typing.cast(PriceNonNull, price)
|
|
|
|
|
|
|
|
- # EXTREME DETAIL: Calculate total daily revenue outputs per metric type.
|
|
|
|
|
m = get_metrics(output['material_amount'] * runs_per_day, price)
|
|
m = get_metrics(output['material_amount'] * runs_per_day, price)
|
|
|
for k in revenue: revenue[k] += m[k]
|
|
for k in revenue: revenue[k] += m[k]
|
|
|
outputs.append(MatPrice(output['material_ticker'], output['material_amount'], price.vwap_7d, price.bid, price.ask))
|
|
outputs.append(MatPrice(output['material_ticker'], output['material_amount'], price.vwap_7d, price.bid, price.ask))
|
|
@@ -103,15 +99,18 @@ def calc_profit(recipe: Recipe, buildings: typing.Mapping[str, Building], hab_ar
|
|
|
output_per_day = lowest_liquidity['material_amount'] * runs_per_day
|
|
output_per_day = lowest_liquidity['material_amount'] * runs_per_day
|
|
|
|
|
|
|
|
average_traded_7d = output_prices[lowest_liquidity['material_ticker']].average_traded_7d
|
|
average_traded_7d = output_prices[lowest_liquidity['material_ticker']].average_traded_7d
|
|
|
- output_per_area = output_per_day / area
|
|
|
|
|
- market_capacity_area = average_traded_7d / output_per_area
|
|
|
|
|
|
|
+
|
|
|
|
|
+ # EXTREME DETAIL: We convert our metrics from 'per area' to 'per base' (1 base = 500 area).
|
|
|
|
|
+ # We divide the building's area footprint by 500 to determine what fraction of a base it consumes.
|
|
|
|
|
+ output_per_base = output_per_day / (area / 500)
|
|
|
|
|
+ market_capacity_base = average_traded_7d / output_per_base
|
|
|
|
|
|
|
|
- logistics_per_area = max(
|
|
|
|
|
|
|
+ logistics_per_base = max(
|
|
|
sum(materials[input['material_ticker']]['weight'] * input['material_amount'] for input in recipe['inputs']),
|
|
sum(materials[input['material_ticker']]['weight'] * input['material_amount'] for input in recipe['inputs']),
|
|
|
sum(materials[input['material_ticker']]['volume'] * input['material_amount'] for input in recipe['inputs']),
|
|
sum(materials[input['material_ticker']]['volume'] * input['material_amount'] for input in recipe['inputs']),
|
|
|
sum(materials[output['material_ticker']]['weight'] * output['material_amount'] for output in recipe['outputs']),
|
|
sum(materials[output['material_ticker']]['weight'] * output['material_amount'] for output in recipe['outputs']),
|
|
|
sum(materials[output['material_ticker']]['volume'] * output['material_amount'] for output in recipe['outputs']),
|
|
sum(materials[output['material_ticker']]['volume'] * output['material_amount'] for output in recipe['outputs']),
|
|
|
- ) * runs_per_day / area
|
|
|
|
|
|
|
+ ) * runs_per_day / (area / 500)
|
|
|
|
|
|
|
|
return Profit(outputs, recipe['recipe_name'],
|
|
return Profit(outputs, recipe['recipe_name'],
|
|
|
expertise=building['expertise'],
|
|
expertise=building['expertise'],
|
|
@@ -122,10 +121,10 @@ def calc_profit(recipe: Recipe, buildings: typing.Mapping[str, Building], hab_ar
|
|
|
revenue=revenue,
|
|
revenue=revenue,
|
|
|
input_costs=input_costs,
|
|
input_costs=input_costs,
|
|
|
runs_per_day=runs_per_day,
|
|
runs_per_day=runs_per_day,
|
|
|
- logistics_per_area=logistics_per_area,
|
|
|
|
|
|
|
+ logistics_per_base=logistics_per_base,
|
|
|
output_per_day=output_per_day,
|
|
output_per_day=output_per_day,
|
|
|
average_traded_7d=average_traded_7d,
|
|
average_traded_7d=average_traded_7d,
|
|
|
- market_capacity_area=market_capacity_area)
|
|
|
|
|
|
|
+ market_capacity_base=market_capacity_base)
|
|
|
|
|
|
|
|
def building_construction_cost(building: Building, prices: typing.Mapping[str, Price]) -> dict[str, float]:
|
|
def building_construction_cost(building: Building, prices: typing.Mapping[str, Price]) -> dict[str, float]:
|
|
|
cost = {'vwap': 0.0, 'bid': 0.0, 'ask': 0.0}
|
|
cost = {'vwap': 0.0, 'bid': 0.0, 'ask': 0.0}
|
|
@@ -192,8 +191,8 @@ class RawPrice(typing.TypedDict):
|
|
|
VWAP7D: float | None
|
|
VWAP7D: float | None
|
|
|
AverageTraded7D: float | None
|
|
AverageTraded7D: float | None
|
|
|
VWAP30D: float | None
|
|
VWAP30D: float | None
|
|
|
- Bid: float | None # Added Bid extraction
|
|
|
|
|
- Ask: float | None # Added Ask extraction
|
|
|
|
|
|
|
+ Bid: float | None
|
|
|
|
|
+ Ask: float | None
|
|
|
|
|
|
|
|
@dataclasses.dataclass(eq=False, frozen=True, slots=True)
|
|
@dataclasses.dataclass(eq=False, frozen=True, slots=True)
|
|
|
class Price:
|
|
class Price:
|
|
@@ -215,21 +214,17 @@ class Profit:
|
|
|
expertise: str
|
|
expertise: str
|
|
|
building: str
|
|
building: str
|
|
|
area: float
|
|
area: float
|
|
|
- capex: dict[str, float] # Transformed from float to Dict
|
|
|
|
|
- opex: dict[str, float] # Transformed from float to Dict
|
|
|
|
|
- revenue: dict[str, float] # Transformed from float to Dict
|
|
|
|
|
|
|
+ capex: dict[str, float]
|
|
|
|
|
+ opex: dict[str, float]
|
|
|
|
|
+ revenue: dict[str, float]
|
|
|
input_costs: typing.Collection[MatPrice]
|
|
input_costs: typing.Collection[MatPrice]
|
|
|
runs_per_day: float
|
|
runs_per_day: float
|
|
|
- logistics_per_area: float
|
|
|
|
|
|
|
+ logistics_per_base: float # Renamed from area to base
|
|
|
output_per_day: float
|
|
output_per_day: float
|
|
|
average_traded_7d: float
|
|
average_traded_7d: float
|
|
|
- market_capacity_area: float
|
|
|
|
|
|
|
+ market_capacity_base: float # Renamed from area to base
|
|
|
|
|
|
|
|
def __lt__(self, other: Profit) -> bool:
|
|
def __lt__(self, other: Profit) -> bool:
|
|
|
- # EXTREME DETAIL: We establish a baseline VWAP sort for the raw JSON payload.
|
|
|
|
|
- # Even though the frontend now dynamically resorts based on UI dropdown permutations,
|
|
|
|
|
- # sorting the initial JSON correctly saves the client from experiencing a 'pop-in' rearrangement
|
|
|
|
|
- # on their very first page load.
|
|
|
|
|
p_a = self.revenue['vwap'] - self.opex['vwap']
|
|
p_a = self.revenue['vwap'] - self.opex['vwap']
|
|
|
be_a = (self.capex['vwap'] + 3 * self.opex['vwap']) / p_a if p_a > 0 else 10000 - p_a
|
|
be_a = (self.capex['vwap'] + 3 * self.opex['vwap']) / p_a if p_a > 0 else 10000 - p_a
|
|
|
|
|
|