EVE 工业助手

Appearance

services.scoring_service ​

源文件 services/scoring_service.py · 由 scripts/gen_api_docs.py 自动生成，请勿手改

模块说明：

评分服务 — 单数据源：ScoringCache, 定价查询, 评分逻辑

包含:

  • ScoringCache：线程安全、有界 TTL 缓存
  • 模块级便利函数：get_price / get_volume / get_system_cost_index / calc_*_score （评分链路取价直查 market.db market_prices，不经 PricingService）
  • ScoringService：可注入的评分服务类（同接口 + calc_reaction_score）；calc_*_score 为薄委托 → scoring_facade 编排（读 DB/缓存）→ domain.scoring 纯算法 被 industry/trade 页各评分 Worker 消费。

函数 ​

_hub_to_system_id ​

python
def _hub_to_system_id(hub: str) -> int | None

将贸易中心名称映射为太阳系 ID。

定义行：39

_sum_materials ​

python
def _sum_materials(per_line: list[dict], runs: int=1, structure_mat_saving: float=1.0) -> list[dict]

各并行线的单轮材料明细按 type_id 合并（qty 求和、subtotal 重算）。

定义行：44

_batch_materials ​

python
def _batch_materials(materials: list[dict], total_runs: int, me_level: int, saving: float) -> list[dict]

给材料明细列表加一份 total_qty（整批取整），返回新列表。

定义行：76

_default_db ​

python
def _default_db() -> DatabaseManager

惰性获取 DatabaseManager（经容器，消除模块级单例双轨）。

定义行：89

_unit_cost_with_overrides ​

python
def _unit_cost_with_overrides(result: dict, cost_overrides: dict[int, float], base: float) -> float

把「同组自制件按自制价计」折算进单件自制成本。

定义行：94

cache_key ​

python
def cache_key(type_id: int, mode: str, hub: str, char_name: str) -> str

⚠️ 待补 docstring

此函数暂无 docstring，欢迎补充。

定义行：126

invalidate_cache ​

python
def invalidate_cache()

清空模块级研究成本缓存（价格刷新后调用）。

定义行：133

get_price ​

python
def get_price(type_id: int, price_type: str, hub: str | None=None, _db: DatabaseManager | None=None) -> float | None

从 market_prices 获取指定区域的价格。price_type: 'buy' / 'sell'。

定义行：152

get_volume ​

python
def get_volume(type_id: int, vol_type: str='total', hub: str | None=None, _db: DatabaseManager | None=None) -> int

获取指定区域的成交量。vol_type: 'buy' / 'sell' / 'total'

定义行：162

get_system_cost_index ​

python
def get_system_cost_index(system_id: int | None, activity: str='manufacturing', _db: DatabaseManager | None=None, hub: str='Jita') -> float

星系的制造成本指数。system_id=None 时从 hub 推断。

定义行：172

get_adjusted_price ​

python
def get_adjusted_price(type_id: int, _db: DatabaseManager | None=None) -> float | None

获取 ESI adjusted price（EIV 计算用）。兜底 None → 用 sell_price。

定义行：184

_research_cost_cached ​

python
def _research_cost_cached(_db: DatabaseManager, type_id: int, *, solar_system_id: int | None=None) -> float

按 type_id + 设施星系计算研究成本（拷贝/发明），带进程内缓存；失败返回 0。

定义行：197

_clear_research_cost_cache ​

python
def _clear_research_cost_cache() -> None

清空研究成本缓存（价格刷新时调用）。

定义行：221

_empty_plan_metrics ​

python
def _empty_plan_metrics() -> dict

计划指标的零值骨架（键与制造路径一致，供失败分支直接返回）。

定义行：231

_resolve_blueprint_for_product ​

python
def _resolve_blueprint_for_product(db, product_type_id) -> int

按产物 type_id 反查它的制造蓝图（旧计划行只带 product_type_id 时用）。缺失 → 0。

定义行：249

materials_with_names ​

python
def materials_with_names(db, mats: list[tuple[int, int]], prices: dict[int, float]) -> list[dict]

[(type_id, qty)] + 单价 → [{type_id, name, qty, unit_price}]（供个人利润率/明细展示）。

定义行：270

类 ​

class ScoringService ​

⚠️ 待补 docstring

此类暂无 docstring，欢迎补充。

定义行：304

方法 ​

__init__ ​
python
def __init__(self, db: DatabaseManager, cache: TtlLRUCache, char_config: dict | None=None)

⚠️ 待补 docstring

此函数暂无 docstring，欢迎补充。

定义行：305

invalidate_cache ​
python
def invalidate_cache(self) -> None

清空评分缓存（价格刷新后调用，避免旧价格评分被复用）

定义行：310

_calc_broker_rate ​
python
def _calc_broker_rate(self, skills: dict, market_data: dict) -> float

⚠️ 待补 docstring

此函数暂无 docstring，欢迎补充。

定义行：318

_calc_relist_discount ​
python
def _calc_relist_discount(self, skills: dict) -> float

⚠️ 待补 docstring

此函数暂无 docstring，欢迎补充。

定义行：321

_calc_sales_tax_rate ​
python
def _calc_sales_tax_rate(self, skills: dict) -> float

⚠️ 待补 docstring

此函数暂无 docstring，欢迎补充。

定义行：324

calculate_total_metrics ​
python
def calculate_total_metrics(per_run: dict, runs: int=1, parallels: int=1) -> dict

将 per-run 评分结果按 runs/parallels 缩放到计划总数值。

定义行：330

combine_per_line ​
python
def combine_per_line(per_line: list[dict], runs: int=1) -> tuple[dict, dict]

逐线结果 → 计划级 (per_run, total)。

定义行：396

calculate_plan_metrics ​
python
def calculate_plan_metrics(plan_data: dict, char_config: dict, *, mat_hub: str | None=None, sell_hub: str | None=None, price_type_mat: str | None=None, price_type_prod: str | None=None, system_id: int | None=None, mat_mult: float=1.0, prod_mult: float=1.0) -> dict

从一条生产计划数据计算所有派生指标。

定义行：453

_research_skill_levels ​
python
def _research_skill_levels(conn, blueprint_type_id: int, activity: str, skills: dict) -> tuple[int, int, int, str]

从 blueprint_skills 解析该活动的技能要求并读角色等级。

定义行：689

_calculate_research_metrics ​
python
def _calculate_research_metrics(plan_data: dict, char_config: dict, *, activity: str, mat_hub: str, sell_hub: str, price_type_mat: str, price_type_prod: str, system_id: int | None, effective_system_id: int | None, structure_cost_mult: float, structure_time_mod: float, fac_tax: float, runs: int, parallels: int) -> dict

拷贝/发明/研究作业的指标（编排：查蓝图/SDE → 调 services.plan_metrics 纯函数）。

定义行：729

calculate_personal_margin ​
python
def calculate_personal_margin(result: dict, inv_map: dict[int, tuple[int, float]], runs: int=1, parallels: int=1, cost_overrides: dict[int, SubitemCost] | None=None) -> float

计算考虑库存成本的个人利润率（%）。实现见 services.plan_metrics。

定义行：1043

child_manufacturing_cost ​
python
def child_manufacturing_cost(plan: dict, metrics: dict) -> float

一条子项产线的总制造价 = 材料成本 + 制造作业费。实现见 services.plan_metrics。

定义行：1056

adjust_mother_metrics ​
python
def adjust_mother_metrics(metrics: dict, sub_cost_map: dict[int, SubitemCost], total_mult: int) -> tuple[float, float, float, dict[int, SubitemCost]]

把拆解母项的自制子项按其制造价计入成本。实现见 services.plan_metrics。

定义行：1063

calc_manufacturing_score ​
python
def calc_manufacturing_score(self, type_id: int, char_config: dict, mat_source_hub: str='Jita', sell_hub: str='Jita', facility_tax_pct: float=0.0, price_type_mat: str='sell', price_type_prod: str='sell', bp_me: int=0, bp_te: int=0, system_id: int | None=None, structure_bonus: float=0.0, structure_time_mod: float=1.0, structure_mat_saving: float=1.0, is_alpha: bool=False, mat_price_mult: float=1.0, prod_price_mult: float=1.0, research_costs: dict[int, float | None] | None=None) -> dict

计算制造评分。

定义行：1075

manufacturing_unit_costs ​
python
def manufacturing_unit_costs(self, type_ids: list[int] | set[int], *, mat_hub: str='Jita', price_type_mat: str='sell', mat_mult: float=1.0, char_config: dict | None=None, system_id: int | None=None, facility_tax_pct: float=0.0, hangar_id: int | None=None, cost_overrides: dict[int, float] | None=None) -> dict[int, float]

这些物品自己造一件要多少钱 → {type_id: 单件自制成本}（算不出的物品不在结果里）。

定义行：1129

_reaction_product_ids ​
python
def _reaction_product_ids(self, type_ids: list[int]) -> set[int]

这批里哪些是反应产物（产物只挂在 activity='reaction' 上）—— 一次查询。

定义行：1228

_best_blueprint_levels ​
python
def _best_blueprint_levels(self, type_ids: list[int]) -> dict[int, tuple[int, int]]

{物品 type_id: (ME, TE)} —— 用户库存里「该物品的制造蓝图」最好的那一张。

定义行：1247

calc_trade_score ​
python
def calc_trade_score(self, type_id: int, buy_hub: str='Jita', sell_hub: str='Jita', buy_price_type: str='buy', sell_price_type: str='sell', char_config: dict | None=None, quantity: int=1) -> dict

计算贸易评分。纯算法在 domain.scoring，编排在 services.scoring_facade。

定义行：1283

calc_reaction_score ​
python
def calc_reaction_score(self, type_id: int, char_config: dict, mat_source_hub: str='Jita', sell_hub: str='Jita', facility_tax_pct: float=0.0, price_type_mat: str='sell', price_type_prod: str='sell', system_id: int | None=None, structure_bonus: float=0.0) -> dict

计算反应（Reaction）利润评分。纯算法在 domain.scoring，编排在 services.scoring_facade。

定义行：1310