Files
material-api/web/calculators/__init__.py
2025-06-05 18:57:35 +08:00

298 lines
11 KiB
Python

from decimal import Decimal, ROUND_HALF_UP
from commons.models.data_fujian import DataFujian
from commons.models.data_network import DataNetwork
from commons.models.fujian_survey import FujianSurvey
from commons.models.fuzhou_highway_bureau import FuzhouHighwayBureau
from commons.models.fuzhou_transportation_bureau import FuzhouTransportationBureau
from commons.models.price_result import PriceResult
from commons.models.sanming_steel import SanmingSteel
class Helper:
@staticmethod
def get_last_month(year, month):
if month == 1:
last_month_year = year - 1
last_month = 12
else:
last_month_year = year
last_month = month - 1
return last_month_year, last_month
@staticmethod
def get_next_month(year, month):
if month == 12:
last_month_year = year + 1
last_month = 1
else:
last_month_year = year
last_month = month + 1
return last_month_year, last_month
class RoundMethod:
@staticmethod
def normal(n, round_bit):
q_ext = Decimal('0.' + '0' * round_bit)
result = Decimal(n).quantize(q_ext, rounding=ROUND_HALF_UP)
return result if round_bit != 0 else int(result)
@staticmethod
def none(n, round_bit):
result = round(int(n * 10 ** round_bit) / 10 ** round_bit, round_bit)
return result if round_bit != 0 else int(result)
@staticmethod
def five(n, round_bit):
result = round(int(n * 10 ** round_bit / 5) * 5 / 10**round_bit, round_bit)
return result if round_bit != 0 else int(result)
class Calculator:
material_id = ""
name = ""
year = 0
month = 0
price_ftb = 0
fluctuating_ftb = 0
price_ss = 0
fluctuating_ss = 0
price_fhb = 0
fluctuating_fhb = 0
price_network = 0
fluctuating_network = 0
price_survey = 0
fluctuating_survey = 0
price_last_month = 0
price_calculate = 0
price_recommend = 0
fluctuating_recommend = 0
price_fujian = 0
fluctuating_fujian = 0
weight_ftb = 1
weight_ss = 1
weight_fhb = 1
weight_network = 1
weight_survey = 1
unit = "" # 单位
spec = '' # 规格
display_digit = ''
round_bit = 1 # 结果保留位数
round_method = RoundMethod.normal # 舍入方法
_previous_prices = None # 上期价格数据
_fluctuatings = [] # 浮动
def __init__(self, year, month, force=True):
self.year = year
self.month = month
self.force = True # todo-2 已修改的的发布价不在覆盖计算
@property
def previous_prices(self):
if not self._previous_prices:
previous_price = PriceResult.get_by_key(self.material_id, *Helper.get_last_month(self.year, self.month))
self._previous_prices = previous_price
return self._previous_prices
def result(self):
return {
'material_id': self.material_id,
'name': self.name,
'year': self.year,
'month': self.month,
'price_ftb': self.price_ftb,
'fluctuating_ftb': self.fluctuating_ftb,
'price_ss': self.price_ss,
'fluctuating_ss': self.fluctuating_ss,
'price_fhb': self.price_fhb,
'fluctuating_fhb': self.fluctuating_fhb,
'price_network': self.price_network,
'fluctuating_network': self.fluctuating_network,
'price_survey': self.price_survey,
'fluctuating_survey': self.fluctuating_survey,
'price_last_month': self.price_last_month,
'price_calculate': self.price_calculate,
'price_recommend': self.price_recommend,
'fluctuating_recommend': self.fluctuating_recommend,
'price_fujian': self.price_fujian,
'fluctuating_fujian': self.fluctuating_fujian,
'weight_ftb': self.weight_ftb,
'weight_ss': self.weight_ss,
'weight_fhb': self.weight_fhb,
'weight_network': self.weight_network,
'weight_survey': self.weight_survey,
'unit': self.unit,
'spec': self.spec,
'display_digit': self.display_digit,
}
def run(self):
self._get_weight()
self.price_ftb, self.fluctuating_ftb = self._get_ftb_price()
self.price_ss, self.fluctuating_ss = self._get_ss_price()
self.price_fhb, self.fluctuating_fhb = self._get_fhb_price()
self.price_network, self.fluctuating_network = self._get_network_price()
self.price_survey, self.fluctuating_survey = self._get_survey_price()
self.price_last_month = self._get_last_month_price()
self.price_calculate = self._get_calculate_price()
self.price_recommend, self.fluctuating_recommend = self._get_recommend_price()
self.price_fujian, self.fluctuating_fujian = self._get_fujian_price()
return self
def _get_ftb_price(self):
query = FuzhouTransportationBureau.get_query(self.year, self.month, material_id=self.material_id)
data = query.first()
price = data.price if data else 0
fluctuating = self._get_fluctuating('price_ftb', price)
return price, fluctuating
def _get_ss_price(self):
query = SanmingSteel.get_query(self.year, self.month, material=self.material_id)
data = query.first()
price = data.price if data else 0
fluctuating = self._get_fluctuating('price_ss', price)
return price, fluctuating
def _get_fhb_price(self):
query = FuzhouHighwayBureau.get_query(self.year, self.month, material_id=self.material_id)
data = query.first()
price = data.price if data else 0
fluctuating = self._get_fluctuating('price_fhb', price)
return price, fluctuating
def _get_network_price(self):
query = DataNetwork.get_query(self.year, self.month, material_id=self.material_id)
data = query.first()
price = data.price if data else 0
fluctuating = self._get_fluctuating('price_network', price)
return price, fluctuating
def _get_survey_price(self):
query = FujianSurvey.get_query(self.year, self.month, material_id=self.material_id)
data = query.first()
price = data.price if data else 0
fluctuating = self._get_fluctuating('price_survey', price)
return price, fluctuating
def _get_fujian_price(self):
query = DataFujian.get_query(self.year, self.month, number=self.material_id)
data = query.first()
price = data.price if data else 0
fluctuating = self._get_fluctuating('price_fujian', price)
return price, fluctuating
def _get_last_month_price(self):
return getattr(self.previous_prices, 'price_recommend', 0)
def _get_calculate_price(self, round_dit=2):
prices = [
self.price_ftb * self.weight_ftb,
self.price_ss * self.weight_ss,
self.price_fhb * self.weight_fhb,
self.price_network * self.weight_network,
self.price_survey * self.weight_survey,
]
if sum(prices) == 0:
return 0
result = sum(prices) / sum([self.weight_ftb, self.weight_ss, self.weight_fhb, self.weight_network, self.weight_survey])
result = self.round_method(result, self.round_bit)
return result
def _get_recommend_price(self, round_by=10):
if not self.previous_prices:
return 0, 0
previous_price = float(getattr(self.previous_prices, 'price_recommend', 0))
if not previous_price:
previous_price = float(getattr(self.previous_prices, 'price_calculate', 0))
self._fluctuatings = [
self.fluctuating_ftb * self.weight_ftb,
self.fluctuating_ss * self.weight_ss,
self.fluctuating_fhb * self.weight_fhb,
self.fluctuating_network * self.weight_network,
self.fluctuating_survey * self.weight_survey,
]
fluctuating = sum(self._fluctuatings) / sum([self.weight_ftb, self.weight_ss, self.weight_fhb, self.weight_network, self.weight_survey])
fluctuating = self.round_method(fluctuating, self.round_bit)
price = Decimal(fluctuating) + Decimal(previous_price)
price = self.round_method(price, self.round_bit)
return price, fluctuating
def _get_weight(self):
# 如果有上月数据按上月权重计算 无则平均数
if self.previous_prices:
self.weight_ftb = self._previous_prices.weight_ftb
self.weight_ss = self._previous_prices.weight_ss
self.weight_fhb = self._previous_prices.weight_fhb
self.weight_network = self._previous_prices.weight_network
self.weight_survey = self._previous_prices.weight_survey
self.display_digit = self._previous_prices.display_digit
else:
self.weight_ftb = 1
self.weight_ss = 1
self.weight_fhb = 1
self.weight_network = 1
self.weight_survey = 1
self.display_digit = 1
def save(self):
result = self.result()
PriceResult(**result).upsert()
def _get_fluctuating(self, field_name, price):
previous_price = getattr(self.previous_prices, field_name, 0) or price
fluctuating = price - previous_price if price else 0
return round(fluctuating, 2)
if __name__ == '__main__':
from calculators.asphalt_domestic import AsphaltDomesticCalculator
from calculators.asphalt_imported import AsphaltImportedCalculator
from calculators.cement_325 import Cement325Calculator
from calculators.cement_425 import Cement425Calculator
from calculators.oil_0 import Oil0Calculator
from calculators.oil_89 import Oil89Calculator
from calculators.oil_92 import Oil92Calculator
from calculators.steel_plate import SteelPlateCalculator
from calculators.steel_rebar_300 import Reber300Calculator
from calculators.steel_rebar_400 import Reber400Calculator
from calculators.steel_section import SteelSectionCalculator
from calculators.steel_strand import SteelStrandCalculator
from calculators.asphalt_domestic_modifier import AsphaltDomesticModifierCalculator
from calculators.asphalt_imported_modifier import AsphaltImportedModifierCalculator
from core.factory import ClientApp
for Calculator in [
AsphaltDomesticCalculator,
AsphaltImportedCalculator,
Cement325Calculator,
Cement425Calculator,
Oil0Calculator,
Oil89Calculator,
Oil92Calculator,
SteelPlateCalculator,
Reber300Calculator,
Reber400Calculator,
SteelSectionCalculator,
SteelStrandCalculator,
AsphaltDomesticModifierCalculator,
AsphaltImportedModifierCalculator,
]:
with ClientApp().app_context():
for month in range(8, 13):
calculator = Calculator(year=2023, month=month)
_result = calculator.run()
calculator.save()
print(_result)