257 lines
11 KiB
Python
257 lines
11 KiB
Python
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
|
||
from app.utils.universal_api_manager import universal_api
|
||
|
||
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
|
||
|
||
# 获取专利信息 TODO 参数
|
||
try:
|
||
patent_data = universal_api.query_patent_info("未找到 企业名称、企业统代、企业注册号")
|
||
# 解析专利数量
|
||
patentData = patent_data.get("data", {"dataList":[]})
|
||
patent_data_num = len(patentData["dataList"])
|
||
|
||
# 查询专利是否存在
|
||
patent_data_al = list(filter(lambda x: x.get("SQH") == data.patent_application_no, patentData))
|
||
if len(patent_data_al) > 0:
|
||
# 验证 专利剩余年限
|
||
# TODO 无法验证 专利剩余保护期>10年(10分)，5-10年(7分)，<5年(3分)；
|
||
|
||
# 发展潜力D相关参数 专利数量
|
||
patent_count = calculate_patent_usage_score(len(patent_data_al))
|
||
input_data_by_b1["patent_count"] = patent_count
|
||
else:
|
||
input_data_by_b1["patent_count"] = 0
|
||
except Exception as e:
|
||
input_data_by_b1["patent_count"] = 0
|
||
|
||
# 提取 文化价值B2 计算参数
|
||
input_data_by_b2 = _extract_calculation_params_b2(data)
|
||
# 提取 风险调整系数B3 计算参数
|
||
input_data_by_b3 = _extract_calculation_params_b3(data)
|
||
# 提取 市场估值C 参数
|
||
input_data_by_c = _extract_calculation_params_c(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)
|
||
|
||
# 侵权分 默认 6 TODO 需要使用第三方API 无侵权记录(10分)，历史侵权已解决(6分)，现存纠纷(2分)
|
||
infringement_score = calculate_infringement_score("无侵权信息")
|
||
infringement_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
|
||
baidu_index = "暂无"
|
||
wechat_index = universal_api.wx_index(data.asset_name) # 通过资产信息获取微信指数 TODO 这里返回的没确认指数参数，有可能返回的图示是指数信息
|
||
weibo_index = "暂无"
|
||
search_index_s1 = calculate_search_index_s1(baidu_index,wechat_index,weibo_index) # 默认值，实际应从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,
|
||
'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\bilibili\kuaishou
|
||
douyin_views = universal_api.video_views("douyin",video_id="视频id未知")
|
||
kuaishou_views= universal_api.video_views("kuaishou",video_id="视频id未知")
|
||
bilibili_views= universal_api.video_views("bilibili",video_id="视频id未知")
|
||
# 跨界合作深度 品牌联名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,
|
||
}
|
||
|
||
|
||
# 获取 市场估值C 相关参数
|
||
def _extract_calculation_params_c(data: UserValuationCreate) -> Dict[str, Any]:
|
||
|
||
# 稀缺性乘数C3 发行量
|
||
circulation = data.circulation\
|
||
|
||
# 时效性衰减C4 参数 用户选择距离最近一次市场活动（交易、报价、评估）的相距时间
|
||
last_market_activity = data.last_market_activity
|