from __future__ import annotations from datetime import date, datetime import json from pathlib import Path from threading import RLock from uuid import uuid4 from zoneinfo import ZoneInfo from rdflib import Graph, Literal, Namespace, RDF, URIRef, XSD from app.market.filters import rejection_reasons from app.market.models import CandidateRecord, InquiryRequest from app.market.pricing import build_prices from app.market.scoring import CustomerSignals, score_candidates from app.repository import RDFRepository GESLI = Namespace("https://gesli.example/ontology#") RUN = Namespace("https://gesli.example/recommendation/") class MarketService: def __init__(self, root: Path, repository: RDFRepository): self.root = root self.repository = repository self.config = json.loads( (root / "config" / "market-strategies.json").read_text(encoding="utf-8") ) self.runs: dict[str, dict] = {} self.audit_graph = Graph() self._lock = RLock() def customers(self) -> list[dict]: rows = self.repository.query_file( self.root / "queries" / "market-customers.rq" )["rows"] return [ { "id": row["customer"], "name": row["customerName"], "industry": row.get("industry", ""), "preferred_route": row.get("routeName", ""), "repeat_purchase_count": _as_int(row.get("repeatPurchaseCount")), "price_sensitivity": _as_float(row.get("customerPriceSensitivity"), 0.5), "time_sensitivity": _as_float(row.get("customerTimeSensitivity"), 0.5), "reliability_sensitivity": _as_float( row.get("customerReliabilitySensitivity"), 0.5 ), } for row in rows ] def form_options(self) -> dict: candidates = self._candidate_records() context_rows = self.repository.query_file( self.root / "queries" / "market-context.rq" )["rows"] configured_routes = { (row.get("origin", ""), row.get("destination", "")) for row in context_rows if row.get("kind") == "benchmark" and row.get("origin") and row.get("destination") } origins = sorted({origin for origin, _ in configured_routes}) destinations = sorted({destination for _, destination in configured_routes}) route_types = sorted({ item.route_type for item in candidates if item.route_type and (item.origin, item.destination) in configured_routes }) benchmark_by_route = { (row.get("origin", ""), row.get("destination", "")): _as_float( row.get("amount") ) for row in context_rows if row.get("kind") == "benchmark" } routes = [] for origin, destination in sorted(configured_routes): route_candidates = [ item for item in candidates if item.origin == origin and item.destination == destination and item.departure_date and item.valid_to and item.valid_to >= date(2026, 7, 20) ] departures = [item.departure_date for item in route_candidates] transit_days = [item.transit_days for item in route_candidates] routes.append({ "origin": origin, "destination": destination, "route_types": sorted({ item.route_type for item in route_candidates if item.route_type }), "target_amount": benchmark_by_route.get((origin, destination), 0), "requested_departure_date": ( min(departures).isoformat() if departures else "2026-07-27" ), "max_transit_days": max(transit_days) if transit_days else 30, }) return { "customers": self.customers(), "origins": origins, "destinations": destinations, "route_types": route_types, "routes": routes, "defaults": InquiryRequest(customer="https://gesli.example/sample#Haier").model_dump( mode="json" ), } def recommend(self, inquiry: InquiryRequest) -> dict: customer_row = self._find_customer(inquiry.customer) customer = self._customer_signals(customer_row) component_cost, component_names = self._component_costs( inquiry.evaluation_date, inquiry.origin ) component_product_codes = { code for codes in self.config["component_product_codes_by_origin"].values() for code in codes } route_candidates = [ candidate for candidate in self._candidate_records() if candidate.origin == inquiry.origin and candidate.destination == inquiry.destination and candidate.currency == "CNY" and candidate.product_code not in component_product_codes ] feasible: list[CandidateRecord] = [] rejected: list[dict] = [] for candidate in route_candidates: reasons = rejection_reasons(candidate, inquiry) if reasons: rejected.append(self._rejected_payload(candidate, reasons)) else: feasible.append(candidate) if not feasible: raise ValueError("没有满足当前硬条件的供应商产品和班期") benchmark, strategy = self._market_context(inquiry) scored, weights = score_candidates(feasible, inquiry, customer) candidates: list[dict] = [] for item in scored: candidate: CandidateRecord = item.pop("candidate") total_supplier_cost = candidate.supplier_cost + component_cost prices = build_prices( benchmark_amount=benchmark["amount"], strategy_rate=strategy["adjustment_rate"], solution_type=item["solution_type"], customer=customer, salesperson_adjustment=inquiry.salesperson_adjustment, supplier_cost=total_supplier_cost, config=self.config, ) payload = self._candidate_payload(candidate, item, prices, component_names) candidates.append(payload) top = candidates[0] score_gap = top["total_score"] - candidates[1]["total_score"] if len(candidates) > 1 else 10 confidence = round(min(0.98, 0.78 + max(0, score_gap) / 100), 2) top["confidence_score"] = confidence run_id = f"rec-{uuid4().hex[:12]}" generated_at = datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(timespec="seconds") result = { "run_id": run_id, "generated_at": generated_at, "normalized_inquiry": { **inquiry.model_dump(mode="json"), "customer_name": customer_row["customerName"], "route": f"{inquiry.origin} → {inquiry.destination}", "dangerous_goods_label": "危险品" if inquiry.dangerous_goods else "非危险品", }, "pipeline": { "discovered": len(route_candidates), "rejected": len(rejected), "feasible": len(feasible), "generated_solutions": len(candidates), }, "recommendation": { "candidate_id": top["id"], "solution_type": top["solution_type"], "product_name": top["product_name"], "supplier_name": top["supplier_name"], "final_quote_amount": top["pricing"]["final_quote_amount"], "currency": top["pricing"]["currency"], "confidence_score": confidence, "summary": self._recommendation_summary(top, strategy, customer_row), }, "market_context": { "benchmark_amount": benchmark["amount"], "currency": benchmark["currency"], "strategy_name": strategy["name"], "strategy_objective": strategy["objective"], "strategy_adjustment_rate": strategy["adjustment_rate"], }, "weights": weights, "candidates": candidates, "rejected_candidates": rejected, } self._store_run(run_id, result, inquiry, strategy) return result def get_run(self, run_id: str) -> dict | None: with self._lock: return self.runs.get(run_id) def reset_runtime(self) -> None: with self._lock: self.runs.clear() self.audit_graph = Graph() def get_audit(self, run_id: str) -> dict | None: run = self.get_run(run_id) if not run: return None run_uri = RUN[run_id] with self._lock: triples = list(self.audit_graph.triples((run_uri, None, None))) evaluations = [ evaluation for _, _, evaluation in self.audit_graph.triples( (run_uri, GESLI.hasCandidateEvaluation, None) ) ] for evaluation in evaluations: triples.extend(self.audit_graph.triples((evaluation, None, None))) return { "run_id": run_id, "generated_at": run["generated_at"], "input": run["normalized_inquiry"], "weights": run["weights"], "market_context": run["market_context"], "candidate_count": len(run["candidates"]), "rejected_count": len(run["rejected_candidates"]), "rdf_assertions": len(triples), "sources": self.repository.loaded_file_names, "business_explanation": self._business_audit(run), } def _candidate_records(self) -> list[CandidateRecord]: rows = self.repository.query_file( self.root / "queries" / "market-candidates.rq" )["rows"] return [ CandidateRecord( quote_uri=row["quote"], quote_no=row["quoteNo"], product_uri=row["product"], product_code=row.get("productCode", ""), product_name=row.get("productName", ""), brand_name=row.get("brandName", ""), transport_mode=row.get("transportMode", ""), offering_uri=row["offering"], offering_code=row.get("offeringCode", ""), supplier_uri=row["supplier"], supplier_name=row["supplierName"], schedule_uri=row.get("schedule", ""), schedule_code=row.get("scheduleCode", ""), route_type=row.get("routeType", ""), service_form=row.get("serviceForm", ""), origin=row.get("originLocation", ""), destination=row.get("destinationLocation", ""), currency=row["currency"], supplier_cost=_as_float(row.get("supplierCostAmount")), valid_from=_as_date(row.get("validFrom")), valid_to=_as_date(row.get("validTo")), departure_date=_as_date(row.get("estimatedDepartureTime")), arrival_date=_as_date(row.get("estimatedArrivalTime")), transit_days=_as_float(row.get("estimatedTransitDays")), booking_success_rate=_as_float(row.get("bookingSuccessRate"), 0.8), space_release_success_rate=_as_float( row.get("spaceReleaseSuccessRate"), 0.8 ), etd_on_time_rate=_as_float(row.get("etdAtdOnTimeRate"), 0.85), eta_on_time_rate=_as_float(row.get("etaAtaOnTimeRate"), 0.85), supplier_active=_as_bool(row.get("supplierActive"), True), capacity_available=_as_bool(row.get("capacityAvailable"), True), supports_dangerous_goods=_as_bool( row.get("supportsDangerousGoods"), False ), dangerous_goods_only=_as_bool(row.get("dangerousGoodsOnly"), False), agency_level=row.get("agencyLevel", ""), offering_channel=row.get("offeringChannel", ""), capacity_commitment=row.get("capacityCommitment", ""), spot_space_status=row.get("spotSpaceStatus") or "需确认", spot_space_teu=_as_float(row.get("spotSpaceTeu")), solution_assurance=_as_float(row.get("solutionAssuranceScore"), 0.72), service_support_score=_as_float(row.get("serviceSupportScore"), 0.72), relationship_tier=row.get("relationshipTier") or "基础合作", relationship_strength=_as_float(row.get("relationshipStrength"), 0.60), low_price_access_probability=_as_float( row.get("lowPriceAccessProbability"), 0.50 ), priority_space_probability=_as_float( row.get("prioritySpaceProbability"), 0.60 ), relationship_evidence=( row.get("relationshipEvidence") or "按标准合作机制执行" ), ) for row in rows ] def _find_customer(self, customer_id: str) -> dict: rows = self.repository.query_file( self.root / "queries" / "market-customers.rq" )["rows"] for row in rows: if row["customer"] == customer_id or row["customerName"] == customer_id: return row raise ValueError("客户不存在或没有可用画像") @staticmethod def _customer_signals(row: dict) -> CustomerSignals: service_requirement = "；".join( value for value in ( row.get("notificationRequirement", ""), row.get("documentRequirement", ""), row.get("exceptionResponseRequirement", ""), ) if value ) return CustomerSignals( price_sensitivity=_as_float(row.get("customerPriceSensitivity"), 0.5), time_sensitivity=_as_float(row.get("customerTimeSensitivity"), 0.5), reliability_sensitivity=_as_float( row.get("customerReliabilitySensitivity"), 0.5 ), repeat_purchase_count=_as_int(row.get("repeatPurchaseCount")), overdue_days=_as_int(row.get("overdueDays")), service_requirement=service_requirement, ) def _component_costs( self, evaluation_date: date, origin: str ) -> tuple[float, list[str]]: candidates = self._candidate_records() total = 0.0 names: list[str] = [] product_codes = self.config["component_product_codes_by_origin"].get( origin, self.config["component_product_codes_by_origin"].get("__default__", []), ) for product_code in product_codes: valid = [ item for item in candidates if item.product_code == product_code and (not item.valid_from or item.valid_from <= evaluation_date) and (not item.valid_to or evaluation_date <= item.valid_to) ] if valid: selected = min(valid, key=lambda item: item.supplier_cost) total += selected.supplier_cost names.append(selected.product_name) return total, names def _market_context(self, inquiry: InquiryRequest) -> tuple[dict, dict]: rows = self.repository.query_file( self.root / "queries" / "market-context.rq" )["rows"] applicable = [ row for row in rows if row.get("origin") == inquiry.origin and row.get("destination") == inquiry.destination and _within(inquiry.evaluation_date, row.get("validFrom"), row.get("validTo")) ] benchmark_row = next((row for row in applicable if row["kind"] == "benchmark"), None) strategy_row = next((row for row in applicable if row["kind"] == "strategy"), None) if not benchmark_row or not strategy_row: raise ValueError("当前路线缺少有效的市场基准价格或启用策略") return ( { "uri": benchmark_row["resource"], "amount": _as_float(benchmark_row["amount"]), "currency": benchmark_row["currency"], }, { "uri": strategy_row["resource"], "name": strategy_row["strategyName"], "objective": strategy_row["strategyObjective"], "adjustment_rate": _as_float(strategy_row["strategyAdjustmentRate"]), }, ) @staticmethod def _candidate_payload( candidate: CandidateRecord, scores: dict, prices: dict, component_names: list[str], ) -> dict: reliability = round( (candidate.booking_success_rate + candidate.space_release_success_rate) / 2, 2, ) reasons = [ f"我方与{candidate.supplier_name}为{candidate.relationship_tier}：{candidate.relationship_evidence}", _spot_space_summary(candidate.spot_space_status, candidate.spot_space_teu), f"低价获取概率{candidate.low_price_access_probability:.0%}，优先舱位概率{candidate.priority_space_probability:.0%}", f"供应商订舱与放舱综合成功率{reliability:.0%}", f"预计运输时间{candidate.transit_days:g}天", f"{candidate.route_type or '当前'}路由，{candidate.capacity_commitment or '运力已确认'}", ] focus_reasons = { "方案优选": "方案保障、供应商合作与履约表现综合最优", "关系优选": "船公司或订舱代理的合作优势最突出", "价格参考": "当前可执行候选中采购价格最具参考性", "时效参考": "当前可执行候选中预计运输时间最短", } reasons.insert(0, focus_reasons.get(scores["solution_type"], "可执行的备选运输方案")) return { "id": candidate.quote_uri, "quote_no": candidate.quote_no, "solution_type": scores["solution_type"], "rank": scores["rank"], "brand_name": candidate.brand_name, "product_name": candidate.product_name, "supplier_name": candidate.supplier_name, "offering_channel": candidate.offering_channel, "schedule_code": candidate.schedule_code, "departure_date": candidate.departure_date.isoformat() if candidate.departure_date else "", "arrival_date": candidate.arrival_date.isoformat() if candidate.arrival_date else "", "transit_days": candidate.transit_days, "route_type": candidate.route_type, "capacity_commitment": candidate.capacity_commitment, "components": [*component_names, candidate.product_name], "reliability_rate": reliability, "price_score": scores["price_score"], "transit_score": scores["transit_score"], "supplier_execution_score": scores["supplier_execution_score"], "customer_fit_score": scores["customer_fit_score"], "solution_assurance_score": scores["solution_assurance_score"], "relationship_advantage_score": scores["relationship_advantage_score"], "spot_space_score": scores["spot_space_score"], "total_score": scores["total_score"], "confidence_score": None, "relationship": { "supplier_name": candidate.supplier_name, "supplier_channel": candidate.offering_channel, "tier": candidate.relationship_tier, "strength": candidate.relationship_strength, "low_price_access_probability": candidate.low_price_access_probability, "priority_space_probability": candidate.priority_space_probability, "evidence": candidate.relationship_evidence, }, "spot_space": { "status": candidate.spot_space_status, "teu": candidate.spot_space_teu, "summary": _spot_space_summary( candidate.spot_space_status, candidate.spot_space_teu ), }, "reasons": reasons, "pricing": prices, } @staticmethod def _rejected_payload(candidate: CandidateRecord, reasons: list[str]) -> dict: return { "id": candidate.quote_uri, "quote_no": candidate.quote_no, "product_name": candidate.product_name, "supplier_name": candidate.supplier_name, "schedule_code": candidate.schedule_code, "reasons": reasons, } @staticmethod def _recommendation_summary( candidate: dict, strategy: dict, customer_row: dict ) -> str: return ( f"为{customer_row['customerName']}推荐{candidate['solution_type']}：" f"{candidate['supplier_name']}的{candidate['product_name']}。" f"综合得分{candidate['total_score']:.1f}，采用“{strategy['name']}”，" f"当前{candidate['spot_space']['summary']}，" f"在方案保障、供应商合作、现舱与履约可靠性之间取得当前最优平衡。" ) @staticmethod def _business_audit(run: dict) -> dict: top = run["candidates"][0] alternatives = [] for candidate in run["candidates"][1:]: alternatives.append({ "title": f"{candidate['solution_type']}：{candidate['supplier_name']}", "detail": ( f"{candidate['spot_space']['summary']}；" f"我方与该供应商为{candidate['relationship']['tier']}；" f"预计{candidate['transit_days']:g}天到达。" ), }) rejected = [ f"{candidate['supplier_name']}：{'；'.join(candidate['reasons'])}" for candidate in run["rejected_candidates"][:3] ] return { "decision": ( f"本次为{run['normalized_inquiry']['customer_name']}的" f"{run['normalized_inquiry']['route']}询价，在" f"{len(run['candidates'])}个可执行方案中选择了{top['supplier_name']}。" ), "why_recommended": [ f"{top['spot_space']['summary']}，可直接支持本次订舱安排。", ( f"我方与{top['supplier_name']}为{top['relationship']['tier']}，" f"{top['relationship']['evidence']}。" ), ( f"方案为{top['route_type']}，预计{top['transit_days']:g}天，" f"方案保障得分{top['solution_assurance_score']}分。" ), f"建议向客户报价{top['pricing']['final_quote_amount']:,.0f} {top['pricing']['currency']}。", ], "alternatives": alternatives, "rejected": rejected, "selection_basis": "优先看方案保障、与船公司或订舱代理的合作深度和现舱；再比较履约、客户需求、时效与价格。价格用于利润校验和横向比较，但不单独决定首选。", } def _store_run( self, run_id: str, result: dict, inquiry: InquiryRequest, strategy: dict ) -> None: run_uri = RUN[run_id] generated_at = Literal(result["generated_at"], datatype=XSD.dateTime) with self._lock: self.runs[run_id] = result self.audit_graph.add((run_uri, RDF.type, GESLI.RecommendationRun)) self.audit_graph.add((run_uri, GESLI.generatedAt, generated_at)) self.audit_graph.add((run_uri, GESLI.usesMarketStrategy, URIRef(strategy["uri"]))) self.audit_graph.add((run_uri, GESLI.recommendationReason, Literal(result["recommendation"]["summary"]))) customer_uri = URIRef(inquiry.customer) self.audit_graph.add((run_uri, GESLI.quoteForCustomer, customer_uri)) for candidate in result["candidates"]: evaluation_uri = RUN[f"{run_id}/candidate/{candidate['rank']}"] self.audit_graph.add((evaluation_uri, RDF.type, GESLI.CandidateEvaluation)) self.audit_graph.add((run_uri, GESLI.hasCandidateEvaluation, evaluation_uri)) self.audit_graph.add((evaluation_uri, GESLI.evaluatedQuote, URIRef(candidate["id"]))) self.audit_graph.add((evaluation_uri, GESLI.recommendationRank, Literal(candidate["rank"]))) self.audit_graph.add((evaluation_uri, GESLI.totalScore, Literal(candidate["total_score"]))) self.audit_graph.add(( evaluation_uri, GESLI.solutionAssuranceEvaluationScore, Literal(candidate["solution_assurance_score"]), )) self.audit_graph.add(( evaluation_uri, GESLI.relationshipAdvantageScore, Literal(candidate["relationship_advantage_score"]), )) self.audit_graph.add(( evaluation_uri, GESLI.spotSpaceScore, Literal(candidate["spot_space_score"]), )) top_evaluation = RUN[f"{run_id}/candidate/1"] self.audit_graph.add((run_uri, GESLI.recommendedEvaluation, top_evaluation)) def _as_float(value: str | None, default: float = 0.0) -> float: try: return float(value) if value not in (None, "") else default except (TypeError, ValueError): return default def _spot_space_summary(status: str, teu: float) -> str: status_label = status if status.startswith("现舱") else f"现舱{status}" if teu > 0: return f"{status_label}，当前可订{teu:g} TEU" return f"{status_label}，需向供应商确认" def _as_int(value: str | None, default: int = 0) -> int: try: return int(value) if value not in (None, "") else default except (TypeError, ValueError): return default def _as_bool(value: str | None, default: bool = False) -> bool: if value in (None, ""): return default return str(value).lower() in {"true", "1", "yes"} def _as_date(value: str | None) -> date | None: if not value: return None try: return date.fromisoformat(value[:10]) except ValueError: return None def _within(current: date, start: str | None, end: str | None) -> bool: start_date = _as_date(start) end_date = _as_date(end) return (not start_date or start_date <= current) and (not end_date or current <= end_date)