from fastapi import APIRouter, Depends, HTTPException, status from typing import Optional, List, Dict, Any import json from app.controllers.user_valuation import user_valuation_controller from app.schemas.valuation import ( UserValuationCreate, UserValuationQuery, UserValuationList, UserValuationOut, UserValuationDetail ) from app.schemas.base import Success, SuccessExtra from app.utils.app_user_jwt import get_current_app_user_id from app.utils.calculation_engine.cultural_value_b2 import CulturalValueB2Calculator from app.utils.calculation_engine.economic_value_b1.economic_value_b1 import EconomicValueB1Calculator from app.utils.calculation_engine.economic_value_b1.sub_formulas.basic_value_b11 import calculate_popularity_score, \ calculate_infringement_score, calculate_patent_usage_score from app.utils.calculation_engine.economic_value_b1.sub_formulas.traffic_factor_b12 import calculate_search_index_s1 from app.utils.calculation_engine.risk_adjustment_b3 import RiskAdjustmentB3Calculator from app.models.esg import ESG from app.models.industry import Industry from app.models.policy import Policy app_valuations_router = APIRouter(tags=["用户端估值评估"]) @app_valuations_router.post("/calculation", summary="计算估值") async def calculate_valuation( data: UserValuationCreate, user_id: int = Depends(get_current_app_user_id) ): """ 计算估值评估 根据用户提交的估值评估数据，调用计算引擎进行经济价值B1计算 """ try: calculator_by_b1 = EconomicValueB1Calculator() calculator_by_b2 = CulturalValueB2Calculator() # 风险调整系数B3 calculator_by_b3 = RiskAdjustmentB3Calculator() # 根据行业查询 ESG 基准分（优先用行业名称匹配，如用的是行业代码就把 name 改成 code） esg_obj = await ESG.filter(name=data.industry).first() esg_score = float(esg_obj.number) if esg_obj else 0.0 # 根据行业查询 行业修正系数与ROE industry_obj = await Industry.filter(name=data.industry).first() # roe_score = industry_obj.roe fix_num_score = industry_obj.fix_num # 根据行业查询 政策匹配度 policy_obj = await Policy.filter(name=data.industry).first() policy_match_score = policy_obj.score # 提取 经济价值B1 计算参数 input_data_by_b1 = _extract_calculation_params_b1(data) # ESG关联价值 ESG分 (0-10分) input_data_by_b1["esg_score"] = esg_score # 行业修正系数I input_data_by_b1["industry_coefficient"] = fix_num_score # 政策匹配度 input_data_by_b1["policy_match_score"] = policy_match_score # 提取 文化价值B2 计算参数 input_data_by_b2 = _extract_calculation_params_b2(data) # 提取 风险调整系数B3 计算参数 input_data_by_b3 = _extract_calculation_params_b3(data) # 调用 经济价值B1 计算引擎 calculation_result_by_b1 = calculator_by_b1.calculate_complete_economic_value_b1(input_data_by_b1) # 调用 文化价值B2 计算引擎 calculation_result_by_b2 = calculator_by_b2.calculate_complete_cultural_value_b2(input_data_by_b2) # 调用 风险调整系数B3 计算引擎 calculation_result_by_b3 = calculator_by_b3.calculate_complete_risky_value_b3(input_data_by_b3) # 创建估值评估记录 result = await user_valuation_controller.create_valuation( user_id=user_id, data=data ) # 组装返回 result_dict = json.loads(result.model_dump_json()) result_dict['calculation_result'] = calculation_result return Success(data=result_dict, msg="估值计算完成") except Exception as e: raise HTTPException(status_code=400, detail=f"计算失败: {str(e)}") def _extract_calculation_params_b1(data: UserValuationCreate) -> Dict[str, Any]: """ 从用户提交的数据中提取计算所需的参数 Args: data: 用户提交的估值评估数据 Returns: Dict: 计算所需的参数字典 """ # 基础价值B11相关参数 # 财务价值所需数据 从近三年收益计算 three_year_income = data.three_year_income or [0, 0, 0] # 法律强度L相关参数 # 普及地域分值 默认 7分 popularity_score = calculate_popularity_score(data.application_coverage) # TODO 专利剩余年限【专利证书】 需要使用第三方API patent = data.patent_application_no patent_score = 0 # 侵权分 默认 6 TODO 需要使用第三方API 无侵权记录(10分)，历史侵权已解决(6分)，现存纠纷(2分) # infringement_score = calculate_infringement_score() infringement_score = 0 # 发展潜力D相关参数 # 专利使用量 TODO 需要使用第三方API patent_score = calculate_patent_usage_score(0) # 创新投入比 = (研发费用/营收) * 100 try: rd_investment = float(data.rd_investment or 0) annual_revenue = float(data.annual_revenue or 1) # 避免除零 innovation_ratio = (rd_investment / annual_revenue) * 100 if annual_revenue > 0 else 0 except (ValueError, TypeError): innovation_ratio = 0.0 # # 流量因子B12相关参数 # 近30天搜索指数S1 - 从社交媒体数据计算 TODO 需要使用第三方API search_index_s1 = calculate_search_index_s1() # 默认值，实际应从API获取 # 行业均值S2 TODO 系统内置 未找到相关内容 industry_average_s2 = 0.0 # 社交媒体传播度S3 - TODO 需要使用第三方API,click_count view_count 未找到对应参数 # likes: 点赞数(API获取) # comments: 评论数(API获取) # shares: 转发数(API获取) # followers: 粉丝数 # click_count: 商品链接点击量(用户填写) sales_volume 使用 sales_volume 销售量 # view_count: 内容浏览量(用户填写) link_views # 政策乘数B13相关参数 # 政策契合度评分P - 根据行业和资助情况计算 # 实施阶段 implementation_stage = data.application_maturity # 资金支持 funding_support = data.funding_status return { # 基础价值B11相关参数 'three_year_income': three_year_income, 'patent_score': patent_score, 'popularity_score': popularity_score, 'infringement_score': infringement_score, 'innovation_ratio': innovation_ratio, # 流量因子B12相关参数 'search_index_s1': search_index_s1, 'industry_average_s2': industry_average_s2, # 'social_media_spread_s3': social_media_spread_s3, 'likes':likes, 'comments':comments, 'shares':shares, 'followers':followers, 'click_count': int(data.click_count) or 100, 'view_count': int(data.link_views) or 100, # 政策乘数B13相关参数 'implementation_stage': implementation_stage, 'funding_support':funding_support } # 获取 文化价值B2 相关参数 def _extract_calculation_params_b2(data: UserValuationCreate) -> Dict[str, Any]: """ argrg: data: 用户提交的估值评估数据 retus: Dict: 计算所需的参数字典 """ # 活态传承系数B21 县官参数 inheritor_level_coefficient = data.inheritor_level # 传承人等级 offline_sessions = int(data.offline_activities) #线下传习次数 # 以下调用API douyin_views = "抖音播放量" kuaishou_views= "快手播放量" bilibili_views= "B站播放量" # 跨界合作深度 品牌联名0.3，科技载体0.5，国家外交礼品1.0 cross_border_depth = float(data.cooperation_depth) # 纹样基因值B22相关参数 historical_inheritance = "历史传承度HI信息未找到" # 用户上传资产相关的纹样图片，系数识别处理，得出结构复杂度值 TODO 一下两项未实现配置 识别图片功能 structure_complexity = "结构复杂度SC" normalized_entropy = "归一化信息熵H" return { "inheritor_level_coefficient": inheritor_level_coefficient, "offline_sessions": offline_sessions, "douyin_views": douyin_views, "kuaishou_views": kuaishou_views, "bilibili_views": bilibili_views, "cross_border_depth": cross_border_depth, "historical_inheritance": historical_inheritance, "structure_complexity": structure_complexity, "normalized_entropy": normalized_entropy, } # 获取 文化价值B2 相关参数 def _extract_calculation_params_b3(data: UserValuationCreate) -> Dict[str, Any]: # 过去30天最高价格 过去30天最低价格 TODO 需要根据字样进行切分获取最高价和最低价 转换成 float 类型 highest_price,lowest_price= data.price_fluctuation lawsuit_status = "无诉讼" # 诉讼状态 TODO (API获取) inheritor_ages = data.inheritor_age_count # [45, 60, 75] # 传承人年龄列表 return { "highest_price": highest_price, "lowest_price": lowest_price, "lawsuit_status": lawsuit_status, "inheritor_ages": inheritor_ages, }