From 9c219cda8353b0f4fef1e43355aa84854b138056 Mon Sep 17 00:00:00 2001 From: dubingyan666 Date: Sat, 29 Nov 2025 11:18:32 +0800 Subject: [PATCH] up repo --- app/controllers/valuation.py | 359 ++++++++++++++++++++++++----------- 1 file changed, 246 insertions(+), 113 deletions(-) diff --git a/app/controllers/valuation.py b/app/controllers/valuation.py index 037ac4d..cf6e121 100644 --- a/app/controllers/valuation.py +++ b/app/controllers/valuation.py @@ -23,6 +23,80 @@ class ValuationController: model = ValuationAssessment step_model = ValuationCalculationStep + + # 参数说明映射表：将参数名（英文）映射到中文说明 + PARAM_DESCRIPTIONS = { + # 财务价值相关 + "three_year_income": "近三年收益（万元）", + "annual_revenue_3_years": "近三年收益（万元）", + "financial_value_f": "财务价值F", + + # 法律强度相关 + "patent_score": "专利分", + "popularity_score": "普及分", + "infringement_score": "侵权分", + "legal_strength_l": "法律强度L", + + # 发展潜力相关 + "patent_count": "专利数量", + "esg_score": "ESG分", + "innovation_ratio": "创新投入比", + "development_potential_d": "发展潜力D", + + # 行业系数 + "industry_coefficient": "行业系数I", + "target_industry_roe": "目标行业ROE", + "benchmark_industry_roe": "基准行业ROE", + + # 流量因子相关 + "search_index_s1": "搜索指数S1", + "industry_average_s2": "行业均值S2", + "social_media_spread_s3": "社交媒体传播度S3", + "likes": "点赞数", + "comments": "评论数", + "shares": "转发数", + "sales_volume": "销售量", + "link_views": "链接浏览量", + + # 政策乘数相关 + "implementation_stage": "实施阶段评分", + "funding_support": "资金支持度", + "policy_match_score": "政策匹配度", + + # 文化价值相关 + "inheritor_level_coefficient": "传承人等级系数", + "offline_sessions": "线下传习次数", + "douyin_views": "抖音浏览量", + "bilibili_views": "B站浏览量", + "kuaishou_views": "快手浏览量", + "cross_border_depth": "跨界合作深度", + "historical_inheritance": "历史传承度HI", + "structure_complexity": "结构复杂度SC", + "normalized_entropy": "归一化信息熵H", + + # 风险调整相关 + "highest_price": "最高价格", + "lowest_price": "最低价格", + "inheritor_ages": "传承人年龄列表", + "lawsuit_status": "诉讼状态", + + # 市场估值相关 + "manual_bids": "手动竞价列表", + "expert_valuations": "专家估值列表", + "weighted_average_price": "加权平均价格", + "daily_browse_volume": "日均浏览量", + "collection_count": "收藏数", + "issuance_level": "发行量", + "recent_market_activity": "最近市场活动时间", + + # 动态质押率相关 + "monthly_transaction_amount": "月交易额（万元）", + "monthly_amount": "月交易额（万元）", + "heritage_asset_level": "非遗等级", + "dynamic_pledge_rate": "动态质押率", + "base_pledge_rate": "基础质押率", + "flow_correction": "流量修正系数", + } async def create_calculation_step(self, data: ValuationCalculationStepCreate) -> ValuationCalculationStepOut: """ @@ -223,13 +297,12 @@ class ValuationController: """ 根据估值ID生成计算过程的 Markdown 报告。 - 此方法会查询所有相关的计算步骤，按照公式的层级关系组织， + 此方法会查询所有相关的计算步骤，按照公式顺序组织， 并生成格式化的 Markdown 文档，包含： - - 公式名称和说明 + - 公式名称 - 输入参数 + - 公式文本 - 输出结果 - - 计算状态 - - 错误信息（如果有） Args: valuation_id (int): 估值的唯一标识符。 @@ -245,10 +318,10 @@ class ValuationController: if not valuation: raise ValueError(f"估值记录不存在: {valuation_id}") - # 获取所有计算步骤 + # 获取所有计算步骤，按 step_order 排序 steps = await self.step_model.filter(valuation_id=valuation_id).order_by('step_order') if not steps: - return f"# 估值计算报告\n\n**估值ID**: {valuation_id}\n\n**资产名称**: {valuation.asset_name}\n\n> 暂无计算步骤记录。\n" + return f"# 计算摘要\n\n**估值ID**: {valuation_id}\n\n**资产名称**: {valuation.asset_name}\n\n> 暂无计算步骤记录。\n" # 转换为字典列表，便于处理 steps_data = [] @@ -256,11 +329,8 @@ class ValuationController: step_dict = ValuationCalculationStepOut.model_validate(step).model_dump() steps_data.append(step_dict) - # 构建公式树形结构 - formula_tree = self._build_formula_tree(steps_data) - # 生成 Markdown - markdown = self._generate_markdown(valuation, formula_tree) + markdown = self._generate_markdown(valuation, steps_data) logger.info("calcstep.report_markdown generated valuation_id={} steps_count={}", valuation_id, len(steps_data)) return markdown @@ -327,139 +397,202 @@ class ValuationController: return {'roots': root_nodes, 'all': tree} - def _generate_markdown(self, valuation, formula_tree: Dict) -> str: + def _generate_markdown(self, valuation, steps_data: List[Dict]) -> str: """ 生成 Markdown 格式的报告。 Args: valuation: 估值评估对象。 - formula_tree: 公式树形结构。 + steps_data: 计算步骤列表（已按 step_order 排序）。 Returns: str: Markdown 格式的字符串。 """ lines = [] - # 标题和基本信息 - lines.append("# 估值计算报告") + # 标题 + lines.append("# 计算摘要") lines.append("") - lines.append("## 基本信息") - lines.append("") - lines.append("| 字段 | 值 |") - lines.append("|------|----|") - lines.append(f"| 估值ID | {valuation.id} |") - lines.append(f"| 资产名称 | {valuation.asset_name} |") - lines.append(f"| 所属机构 | {valuation.institution} |") - lines.append(f"| 所属行业 | {valuation.industry} |") - heritage_value = valuation.heritage_level if valuation.heritage_level else "-" - lines.append(f"| 非遗等级 | {heritage_value} |") - created_at_str = valuation.created_at.strftime("%Y-%m-%d %H:%M:%S") if valuation.created_at else "N/A" - lines.append(f"| 创建时间 | {created_at_str} |") lines.append("") - # 计算结果摘要 - if valuation.final_value_ab is not None: - lines.append("## 计算结果摘要") - lines.append("") - lines.append("| 项目 | 数值（万元） |") - lines.append("|------|-------------|") - if valuation.model_value_b is not None: - lines.append(f"| 模型估值B | {valuation.model_value_b:.2f} |") - if valuation.market_value_c is not None: - lines.append(f"| 市场估值C | {valuation.market_value_c:.2f} |") - lines.append(f"| **最终估值AB** | **{valuation.final_value_ab:.2f}** |") - if valuation.dynamic_pledge_rate is not None: - lines.append(f"| 动态质押率 | {valuation.dynamic_pledge_rate:.4f} |") - lines.append("") - - # 详细计算过程 - lines.append("## 详细计算过程") - lines.append("") - - def _format_json_block(value: Any, indent_prefix: str = "") -> List[str]: - """格式化 JSON 代码块""" - json_text = json.dumps(value, ensure_ascii=False, indent=2) - block_lines = [f"{indent_prefix}```json"] - for line in json_text.splitlines(): - block_lines.append(f"{indent_prefix}{line}") - block_lines.append(f"{indent_prefix}```") - return block_lines - - # 递归生成公式树 - heading_levels = ["####", "#####", "######", "######", "######"] - - def render_node(node: Dict, level: int = 0, prefix: str = ""): - step = node['step'] - heading = heading_levels[min(level, len(heading_levels) - 1)] + # 遍历所有步骤，按顺序生成 + for step in steps_data: name = step.get('formula_name', step.get('step_name', '未知')) formula_text = step.get('formula_text', step.get('step_description', '')) - status = step.get('status', 'unknown') input_params = step.get('input_params') output_result = step.get('output_result') - error_message = step.get('error_message') - duration_ms = None - if output_result and isinstance(output_result, dict): - duration_ms = output_result.get('duration_ms') - lines.append(f"{heading} {prefix}{name}") + + # 公式标题（二级标题） + lines.append(f"## {name}") lines.append("") - # 公式说明 - if formula_text: - if level == 0: - lines.append(f"> {formula_text}") - lines.append("") - else: - lines.append("计算公式：") - lines.append(f"`{formula_text}`") - lines.append("") - - # 状态和耗时 - status_label = { - 'processing': '计算中', - 'completed': '已完成', - 'failed': '计算失败' - }.get(status, status) - lines.append(f"**状态**: {status_label}") - if duration_ms is not None: - lines.append(f"**耗时**: {duration_ms}ms") - lines.append("") - - # 输入参数 + # 参数部分 if input_params: - lines.append("**输入参数**:") - lines.extend(_format_json_block(input_params, "")) + lines.append("**参数：**") + lines.append("") + # 格式化参数显示 + param_lines = self._format_params(input_params) + lines.extend(param_lines) lines.append("") - # 输出结果 + # 公式部分 + if formula_text: + lines.append("**公式：**") + lines.append("") + lines.append("```") + lines.append(formula_text) + lines.append("```") + lines.append("") + + # 结果部分 if output_result: - # 移除 duration_ms，因为已经在状态中显示了 - result_display = {k: v for k, v in output_result.items() if k != 'duration_ms'} - if result_display: - lines.append("**输出结果**:") - lines.extend(_format_json_block(result_display, "")) + # 提取主要结果值 + result_value = self._extract_main_result(output_result, name) + if result_value is not None: + lines.append("**结果：**") + lines.append("") + lines.append(f"`{result_value}`") lines.append("") - # 错误信息 - if error_message: - lines.append(f"> ⚠️ **错误**: {error_message}") - lines.append("") - - # 子节点 - children = node.get('children', []) - if children: - for idx, child in enumerate(children, start=1): - lines.append("") - child_prefix = f"{prefix}{idx}." - render_node(child, level + 1, child_prefix) lines.append("") - # 渲染所有根节点 - for idx, root in enumerate(formula_tree['roots'], start=1): - prefix = f"{idx}." - render_node(root, 0, prefix) - return "\n".join(lines) + def _format_params(self, params: Dict[str, Any]) -> List[str]: + """ + 格式化参数显示，优先使用列表格式（如果是数组），否则显示为列表项。 + 参数名会附带中文说明（如果存在）。 + + Args: + params: 参数字典 + + Returns: + List[str]: 格式化后的参数行列表 + """ + lines = [] + + def _get_param_label(key: str) -> str: + """获取参数标签，包含中文说明""" + description = self.PARAM_DESCRIPTIONS.get(key) + if description: + return f"{key}（{description}）" + return key + + # 如果参数只有一个键，且值是数组，直接显示数组（不带参数名，符合示例格式） + if len(params) == 1: + key, value = next(iter(params.items())) + if isinstance(value, (list, tuple)): + # 格式化为列表：- [12.2, 13.2, 14.2] + value_str = json.dumps(list(value), ensure_ascii=False) + lines.append(f"- {value_str}") + return lines + + # 多个参数或非数组，显示为列表项（带说明） + for key, value in params.items(): + param_label = _get_param_label(key) + if isinstance(value, (list, tuple)): + value_str = json.dumps(list(value), ensure_ascii=False) + lines.append(f"- **{param_label}**: {value_str}") + elif isinstance(value, dict): + value_str = json.dumps(value, ensure_ascii=False) + lines.append(f"- **{param_label}**: {value_str}") + else: + lines.append(f"- **{param_label}**: {value}") + + return lines + + def _extract_main_result(self, output_result: Dict[str, Any], formula_name: str) -> Optional[str]: + """ + 从输出结果中提取主要结果值。 + + 优先顺序： + 1. 如果结果中只有一个数值类型的值，返回该值 + 2. 如果结果中包含与公式名称相关的字段（如 "财务价值 F" -> "financial_value_f"），返回该值 + 3. 如果结果中包含常见的计算结果字段（如 "result", "value", "output"），返回该值 + 4. 返回第一个数值类型的值 + + Args: + output_result: 输出结果字典 + formula_name: 公式名称 + + Returns: + Optional[str]: 主要结果值的字符串表示，如果找不到则返回 None + """ + if not output_result or not isinstance(output_result, dict): + return None + + # 移除 duration_ms 等元数据字段 + filtered_result = {k: v for k, v in output_result.items() + if k not in ['duration_ms', 'duration', 'timestamp', 'status']} + + if not filtered_result: + return None + + # 如果只有一个值，直接返回 + if len(filtered_result) == 1: + value = next(iter(filtered_result.values())) + if isinstance(value, (int, float)): + return str(value) + elif isinstance(value, (list, tuple)) and len(value) == 1: + return str(value[0]) + else: + return json.dumps(value, ensure_ascii=False) + + # 尝试根据公式名称匹配字段 + # 例如："财务价值 F" -> 查找 "financial_value_f", "财务价值F" 等 + # 提取公式名称中的关键部分（通常是最后一个字母或单词） + name_parts = formula_name.split() + if name_parts: + # 获取最后一个部分（通常是字母，如 "F", "L", "D"） + last_part = name_parts[-1].lower() + # 构建可能的字段名：如 "financial_value_f", "legal_strength_l" 等 + # 将中文名称转换为可能的英文字段名模式 + possible_keys = [] + + # 1. 直接匹配包含最后部分的字段（如包含 "f", "l", "d"） + for key in filtered_result.keys(): + if last_part in key.lower() or key.lower().endswith(f"_{last_part}"): + possible_keys.append(key) + + # 2. 尝试匹配常见的命名模式 + # 例如："财务价值 F" -> "financial_value_f" + # 这里我们尝试匹配以最后部分结尾的字段 + suffix_patterns = [ + f"_{last_part}", + f"_{last_part}_", + last_part, + ] + + for key in filtered_result.keys(): + key_lower = key.lower() + for pattern in suffix_patterns: + if key_lower.endswith(pattern) or pattern in key_lower: + if key not in possible_keys: + possible_keys.append(key) + + # 按优先级匹配 + for key in possible_keys: + if key in filtered_result: + value = filtered_result[key] + if isinstance(value, (int, float)): + return str(value) + + # 查找常见的结果字段 + common_result_keys = ['result', 'value', 'output', 'final_value', 'calculated_value'] + for key in common_result_keys: + if key in filtered_result: + value = filtered_result[key] + if isinstance(value, (int, float)): + return str(value) + + # 返回第一个数值类型的值 + for key, value in filtered_result.items(): + if isinstance(value, (int, float)): + return str(value) + + # 如果都没有，返回整个结果的 JSON（但简化显示） + return json.dumps(filtered_result, ensure_ascii=False) + async def create(self, data: ValuationAssessmentCreate, user_id: int) -> ValuationAssessmentOut: """创建估值评估""" # 将用户ID添加到数据中