# -*- coding: utf-8 -*- """ 美国售价模型 V2 (shopfad) 通过输入多包裹体积和采购价格，快速计算售价 规则版本: 2604 """ import sys sys.path.append(r'D:\workspace\dags\logistics') import math from utils.gtools import MySQLconnect import pandas as pd class USLogisticsPrice: """美国物流价格查询""" _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._init_data() return cls._instance def _init_data(self): """从数据库加载价格数据""" conn = MySQLconnect('logistics') with conn as c: self.sell_usps = pd.read_sql("SELECT * FROM sell_usps", c.con) # 2604版本价格表 (表名按需修改) self.sell_uandf_2604 = pd.read_sql("SELECT * FROM sell_uandf_2604", c.con) self.sell_fedex_price_ref_2604 = pd.read_sql("SELECT * FROM sell_fedex_price_ref_2604", c.con) self.sell_fedex_order_ref_2604 = pd.read_sql("SELECT * FROM sell_fedex_order_ref_2604", c.con) # 物流比例限制表 (表名按需修改) self.logistics_ratio_limit = pd.read_sql("SELECT * FROM logistics_ratio_limit", c.con) # ========== 包裹计费类型判断 ========== def get_package_types(self, packages): """ USPS：默认 UandF：最长边>50 或 次长边>40 或 最短边>30 或 实重>2718 或 体积重>4500 FEDEX：最长边>264 或 实重>67000 或 体积重>95000 或 围长>391 卡派：同FEDEX条件（或该SKU的其他包裹为卡派类型） """ types = [] for pkg in packages: l, w, h, wt = pkg['length_cm'], pkg['width_cm'], pkg['height_cm'], pkg['weight_g'] sizes = sorted([l, w, h], reverse=True) fst, sed, trd = sizes[0], sizes[1], sizes[2] vol_weight = l * w * h / 6 girth = fst + sed * 2 + trd * 2 pkg_type = 'USPS' if fst > 50 or sed > 40 or trd > 30 or wt > 2718 or vol_weight > 4500: pkg_type = 'UandF' if fst > 264 or wt > 67000 or vol_weight > 95000 or girth > 391: pkg_type = 'FEDEX' types.append(pkg_type) # FEDEX升级为卡派（同SKU联动） if 'FEDEX' in types: types = ['卡派' if t == 'FEDEX' else t for t in types] return types # ========== 快递费计算 ========== def _lookup_express_base_fee(self, pkg_type, lbs_weight, oz_weight, profit_rate, fuel_rate): """ 查表获取单个包裹的快递基础计费 USPS/UandF: 直接查表 FEDEX: FEDEX售价参考表 * (1+快递燃油费率) + 2.9 """ if pkg_type == 'USPS': row = self.sell_usps[self.sell_usps['oz'] == oz_weight] if len(row) > 0: return row['最终费用_v2'].iloc[0] / profit_rate, 'USPS' else: return self._lookup_express_base_fee('UandF', lbs_weight, oz_weight, profit_rate, fuel_rate) if pkg_type == 'UandF': row = self.sell_uandf_2604[self.sell_uandf_2604['lbs'] == lbs_weight] if len(row) > 0: return row['加权价格_v2'].iloc[0] / profit_rate, 'UandF' else: return self._lookup_express_base_fee('FEDEX', lbs_weight, oz_weight, profit_rate, fuel_rate) if pkg_type in ('FEDEX', '卡派'): row = self.sell_fedex_price_ref_2604[self.sell_fedex_price_ref_2604['lbs'] == lbs_weight] if len(row) > 0: return row['售价尾端价格_v4'].iloc[0] * (1 + fuel_rate) + 2.9, 'FEDEX' else: return 99999, 'FEDEX' return 0, pkg_type def _calc_express_fee(self, packages, types, profit_rate, fuel_rate): """ 快递费 = Σ(单包裹快递基础计费 + 大包裹费 + 超重费 + 超尺寸费) 大包裹费: 最长边>238 或 围长>315 → 109.4 超重费: 实重>21000 → 6.1 超尺寸费: 最长边>116 或 围长>251 → 7.4 """ total_fee = 0 for i, pkg in enumerate(packages): l, w, h, wt = pkg['length_cm'], pkg['width_cm'], pkg['height_cm'], pkg['weight_g'] sizes = sorted([l, w, h], reverse=True) fst, sed, trd = sizes[0], sizes[1], sizes[2] volume = l * w * h girth = fst + sed * 2 + trd * 2 lbs_weight = math.ceil(max(volume / 8500 / 0.453, wt / 453)) oz_weight = math.ceil(wt / 28) pkg_type = types[i] base_fee, actual_type = self._lookup_express_base_fee( pkg_type, lbs_weight, oz_weight, profit_rate, fuel_rate ) large_pkg_fee = 0 overweight_fee = 0 oversize_fee = 0 if actual_type in ('FEDEX', '卡派'): if fst > 238 or girth > 315: large_pkg_fee = 109.4 if wt > 21000: overweight_fee = 6.1 if fst > 116 or girth > 251: oversize_fee = 7.4 total_fee += base_fee + large_pkg_fee + overweight_fee + oversize_fee return total_fee # ========== 卡派费计算 ========== def _calc_ltl_fee(self, packages, profit_rate): """ 卡派费 = 卡派基础费 + 超长费 + 超重费 合并体积 = ceil(长x宽x高/1000000 x 35.3) (立方英尺) """ total_cubic_feet = 0 over_length_fee = 0 over_weight_fee = 0 for pkg in packages: l, w, h, wt = pkg['length_cm'], pkg['width_cm'], pkg['height_cm'], pkg['weight_g'] sizes = sorted([l, w, h], reverse=True) fst = sizes[0] volume = l * w * h cubic_feet = math.ceil(volume / 1000000 * 35.3147) total_cubic_feet += cubic_feet # 超长费: 最长边>250 if fst > 250: over_length_fee = max(over_length_fee, round(130 / profit_rate - 118, 1)) # 超重费 if 130000 <= wt < 157000: over_weight_fee = max(over_weight_fee, round(130 / profit_rate - 118, 1)) elif 111000 <= wt < 130000: over_weight_fee = max(over_weight_fee, round(80 / profit_rate - 78, 1)) # 卡派基础费 v = total_cubic_feet if v < 25: base_fee = round(198 / profit_rate / 2, 2) elif v < 35: base_fee = round(231 / profit_rate / 2, 2) else: base_fee = round(max(231, 6.6 * v) / profit_rate / 2, 2) ltl_fee = base_fee + over_length_fee + over_weight_fee return ltl_fee, total_cubic_feet # ========== 初始前端展示物流费 ========== def _calc_display_logistics_fee(self, packages, types, total_cubic_feet): """ 卡派: v<25→90, v<35→90+(v-25)*3, 其他→99+(v-35)*5 快递: FEDEX订单参考表查表 """ has_ltl = '卡派' in types if has_ltl: v = total_cubic_feet if v < 25: display = 90 elif v < 35: display = 90 + (v - 25) * 3 else: display = 99 + (v - 35) * 5 print(f"[卡派展示物流费] 合并体积={v} cu ft, 展示费={display}") return display else: # 快递: FEDEX订单参考表查表 total = 0 for pkg in packages: l, w, h, wt = pkg['length_cm'], pkg['width_cm'], pkg['height_cm'], pkg['weight_g'] volume = l * w * h lbs_weight = math.ceil(max(volume / 8500 / 0.453, wt / 453)) row = self.sell_fedex_order_ref_2604[ self.sell_fedex_order_ref_2604['lbs'] == lbs_weight ] if len(row) > 0: total += row['售价尾端价格_v3'].iloc[0] return total # ========== 物流比例限制查表 ========== def _lookup_logistics_limit(self, category, price): """通过条目和价格查表得到建议物流比例上限和建议物流绝对上限""" try: rows = self.logistics_ratio_limit[ (self.logistics_ratio_limit['条目'] == category) & (self.logistics_ratio_limit['价格'] <= price) ] if len(rows) > 0: row = rows.iloc[-1] return row['建议物流比例上限'], row['建议物流绝对上限'] return 1.0, float('inf') except: return 1.0, float('inf') # ========== 统一尾端费用接口 ========== def get_tail_fee_info(self, packages, profit_rate=0.359, fuel_rate=0): """ 尾端费用 = min(快递费, 卡派费) 存在卡派类型包裹则强制使用卡派 """ types = self.get_package_types(packages) has_ltl = '卡派' in types express_fee = self._calc_express_fee(packages, types, profit_rate, fuel_rate) ltl_fee, total_cubic_feet = self._calc_ltl_fee(packages, profit_rate) if has_ltl: tail_fee = ltl_fee else: tail_fee = min(express_fee, ltl_fee) display_fee = self._calc_display_logistics_fee(packages, types, total_cubic_feet) return { 'tail_fee': tail_fee, 'display_logistics_fee': display_fee, 'express_fee': express_fee, 'ltl_fee': ltl_fee, 'types': types, 'has_ltl': has_ltl, 'total_cubic_feet': total_cubic_feet } class USSellPrice: """ 美国售价计算器 V2 (shopfad) 规则版本: 2604 使用示例: calculator = USSellPrice() result = calculator.calculate( packages=[{'length_cm': 30, 'width_cm': 20, 'height_cm': 10, 'weight_g': 500}], purchase_price_cny=100, preset='shopfad' ) """ PRESETS = { 'shopfad': { 'ocean_first_cny': 0.5, 'ocean_first_usd': 0.7, 'air_first_usd': 0.65, 'air_cny_type': 0.093, 'air_first_fix': 22.7, 'exchange_rate': 6.5, 'profit_rate': 0.38, 'air_rate': 0.7, 'tax_rate': 0.1, 'fuel_rate': 0, 'extra_rate': 0 }, 'shopfad2': { 'ocean_first_cny': 1, 'ocean_first_usd': 1, 'air_first_usd': 0.65, 'air_cny_type': 0.093, 'air_first_fix': 22.7, 'exchange_rate': 6.5, 'profit_rate': 0.38, 'air_rate': 0.7, 'tax_rate': 0.145, 'fuel_rate': 0, 'extra_rate': 0 }, 'shopfad3': { 'ocean_first_cny': 0.67, 'ocean_first_usd': 0.7, 'air_first_usd': 0.65, 'air_cny_type': 0.093, 'air_first_fix': 22.7, 'exchange_rate': 6.5, 'profit_rate': 0.3, 'air_rate': 0.7, 'tax_rate': 0.1, 'fuel_rate': 0, 'extra_rate': 0 }, 'shopfad4': { 'ocean_first_cny': 0.67, 'ocean_first_usd': 0.7, 'air_first_usd': 0.65, 'air_cny_type': 0.093, 'air_first_fix': 22.7, 'exchange_rate': 6.5, 'profit_rate': 0.3, 'air_rate': 0.7, 'tax_rate': 0.1, 'fuel_rate': 0, 'extra_rate': 0 } } def __init__(self): self.logistics = USLogisticsPrice() def calculate(self, packages, purchase_price_cny, preset='shopfad', config=None, category=''): """ 计算售价 Args: packages: 包裹列表 [{'length_cm', 'width_cm', 'height_cm', 'weight_g'}] purchase_price_cny: 采购价(CNY) preset: 预设配置名称 config: 自定义配置 (优先使用) category: 条目 (用于物流比例限制查表) """ cfg = config or self.PRESETS.get(preset, self.PRESETS['shopfad']) # 1. 获取尾端费用 fuel_rate = cfg.get('fuel_rate', 0) tail_info = self.logistics.get_tail_fee_info(packages, cfg['profit_rate'], fuel_rate) tail_fee = tail_info['tail_fee'] display_fee = tail_info['display_logistics_fee'] # 2. 判断SPU运输方式 has_ltl = tail_info['has_ltl'] shipping_type = '海运' if has_ltl else '空运' # 3. 汇总体积重 (长x宽x高/6) total_volume_weight = sum( p['length_cm'] * p['width_cm'] * p['height_cm'] / 6 for p in packages ) tax_amount = purchase_price_cny * cfg['tax_rate'] / cfg['exchange_rate'] # 4. 根据运输方式计算调整前售价 if shipping_type == '海运': pre_adjust_price = self._calc_ocean_price( total_volume_weight, purchase_price_cny, tax_amount, tail_fee, cfg ) first_cny = cfg['ocean_first_cny'] / 1000 * total_volume_weight first_usd = cfg['ocean_first_usd'] / 1000 * total_volume_weight else: pre_adjust_price = self._calc_air_price( packages, purchase_price_cny, tax_amount, tail_fee, cfg ) first_cny, first_usd = self._calc_air_first_leg(packages, cfg) # 5. 最终修正 (仅海运) shared_tail_fee = 0 final_price = pre_adjust_price if shipping_type == '海运': final_price, shared_tail_fee = self._apply_ocean_adjustment( pre_adjust_price, display_fee, category ) # 头程费用 / 总物流费用 head_fee = first_cny / cfg['exchange_rate'] + first_usd total_logistics_fee = head_fee + tail_fee return { 'sell_price_usd': round(final_price, 2), 'sell_price_cny': round(final_price * cfg['exchange_rate'], 2), 'purchase_price_cny': purchase_price_cny, 'express_fee': round(tail_fee, 2), 'first_cny': round(first_cny, 2), 'first_usd': round(first_usd, 2), 'tax_amount': round(tax_amount, 2), 'shipping_type': shipping_type, 'preset': preset, 'pre_adjust_price': round(pre_adjust_price, 2), 'display_logistics_fee': round(display_fee, 2), 'shared_tail_fee': round(shared_tail_fee, 2), 'head_fee': round(head_fee, 2), 'total_logistics_fee': round(total_logistics_fee, 2), 'package_types': tail_info['types'], } # ========== 海运计费 ========== def _calc_ocean_price(self, total_volume_weight, purchase_price_cny, tax_amount, tail_fee, cfg): """ 调整前售价 = (((海运头程CNY总价 + 采购价格) / 汇率 + 海运头程USD总价) + 采购价格 x 税金系数 / 汇率) / 利润系数 + 尾端费用 """ ocean_first_cny_total = cfg['ocean_first_cny'] / 1000 * total_volume_weight ocean_first_usd_total = cfg['ocean_first_usd'] / 1000 * total_volume_weight pre_adjust = ( ((ocean_first_cny_total + purchase_price_cny) / cfg['exchange_rate'] + ocean_first_usd_total + tax_amount) / cfg['profit_rate'] + tail_fee ) return pre_adjust # ========== 空运计费 ========== def _calc_air_price(self, packages, purchase_price_cny, tax_amount, tail_fee, cfg): """ 调整前售价 = (采购价格/汇率/利润系数 + ((空运头程CNY总价/汇率 + 空运头程USD总价 + 税金) / 利润系数 + 尾端费用) x 空运分配比例) x (1 + 附加费率) """ air_cny_total = 0 air_usd_total = 0 for pkg in packages: volume = pkg['length_cm'] * pkg['width_cm'] * pkg['height_cm'] volume_weight = volume / 6 density = pkg['weight_g'] / (volume / 1000) if volume > 0 else 0 type_weight = min(337, max(37, math.ceil(density))) air_cny_unit = cfg['air_cny_type'] / 1000 * type_weight + cfg['air_first_fix'] / 1000 air_cny_total += air_cny_unit * volume_weight air_usd_total += cfg['air_first_usd'] / 1000 * volume_weight purchase_usd = purchase_price_cny / cfg['exchange_rate'] pre_adjust = ( purchase_usd / cfg['profit_rate'] + ((air_cny_total / cfg['exchange_rate'] + air_usd_total + tax_amount) / cfg['profit_rate'] + tail_fee) * cfg['air_rate'] ) * (1 + cfg.get('extra_rate', 0)) return pre_adjust def _calc_air_first_leg(self, packages, cfg): """计算空运头程费用 (CNY/USD)""" first_cny = 0 first_usd = 0 for pkg in packages: volume = pkg['length_cm'] * pkg['width_cm'] * pkg['height_cm'] volume_weight = volume / 6 density = pkg['weight_g'] / (volume / 1000) if volume > 0 else 0 type_weight = min(337, max(37, math.ceil(density))) air_cny_unit = cfg['air_cny_type'] / 1000 * type_weight + cfg['air_first_fix'] / 1000 first_cny += air_cny_unit * volume_weight first_usd += cfg['air_first_usd'] / 1000 * volume_weight return first_cny, first_usd # ========== 最终修正 (海运) ========== def _apply_ocean_adjustment(self, pre_adjust_price, display_fee, category): """ 前端展示物流费 = min(初始前端展示物流费, 调整前售价×比例上限, 绝对上限) if min == 初始前端展示物流费: 分摊尾端=0, 最终售价=调整前售价 else: 分摊尾端=初始前端展示物流费-min, 最终售价=调整前售价+分摊尾端 """ total = display_fee + pre_adjust_price if total <= 0: return pre_adjust_price, 0 ratio_limit, abs_limit = self.logistics._lookup_logistics_limit(category, pre_adjust_price) actual_display_fee = min(display_fee, pre_adjust_price * ratio_limit, abs_limit) if actual_display_fee == display_fee: return pre_adjust_price, 0 else: shared_tail = display_fee - actual_display_fee final_price = pre_adjust_price + shared_tail return final_price, shared_tail if __name__ == '__main__': calc = USSellPrice() print("=== 单包裹测试 ===") result = calc.calculate( packages=[{'length_cm': 30, 'width_cm': 20, 'height_cm': 10, 'weight_g': 500}], purchase_price_cny=100, preset='shopfad' ) print(f" 30x20x10cm 500g => ${result['sell_price_usd']}") print(f" 运输方式: {result['shipping_type']}") print(f" 包裹类型: {result['package_types']}") print("\n=== 多包裹测试 ===") result = calc.calculate( packages=[ {'length_cm': 30, 'width_cm': 20, 'height_cm': 10, 'weight_g': 500}, {'length_cm': 40, 'width_cm': 30, 'height_cm': 15, 'weight_g': 1000}, ], purchase_price_cny=200, preset='shopfad' ) print(f" 两件包裹 => ${result['sell_price_usd']}") print(f" 运输方式: {result['shipping_type']}") print(f" 包裹类型: {result['package_types']}") print(f" 尾端费用: ${result['express_fee']}") print(f" 头程CNY: {result['first_cny']}, 头程USD: ${result['first_usd']}") print(f" 总物流: ${result['total_logistics_fee']}")