zfc/guzhi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

all

Browse Source
This commit is contained in:
dubingyan666 2025-10-06 16:01:58 +08:00
parent 422c0135d1
commit 982da3f076
8 changed files with 513 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

187
app/api/v1/calculation/calcuation.py
View File

@ -1,6 +1,8 @@
from fastapi import APIRouter, Depends, HTTPException, status
from typing import Optional, List, Dict, Any
import json
import asyncio
import time
from app.controllers.user_valuation import user_valuation_controller
from app.schemas.valuation import (
@ -12,12 +14,13 @@ from app.schemas.valuation import (
)
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 import FinalValueACalculator
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.log.log import logger
from app.models.esg import ESG
@ -39,26 +42,34 @@ async def calculate_valuation(
根据用户提交的估值评估数据，调用计算引擎进行经济价值B1计算
"""
try:
calculator_by_b1 = EconomicValueB1Calculator()
calculator_by_b2 = CulturalValueB2Calculator()
# 风险调整系数B3
calculator_by_b3 = RiskAdjustmentB3Calculator()
start_ts = time.monotonic()
logger.info("valuation.calc_start user_id={} asset_name={} industry={}", user_id, getattr(data, 'asset_name', None), getattr(data, 'industry', None))
# 根据行业查询 ESG 基准分（优先用行业名称匹配，如用的是行业代码就把 name 改成 code）
esg_obj = await ESG.filter(name=data.industry).first()
esg_score = float(esg_obj.number) if esg_obj else 0.0
esg_obj = None
industry_obj = None
policy_obj = None
try:
esg_obj = await asyncio.wait_for(ESG.filter(name=data.industry).first(), timeout=2.0)
except Exception as e:
logger.warning("valuation.esg_fetch_timeout industry={} err={}", data.industry, repr(e))
esg_score = float(getattr(esg_obj, 'number', 0.0) or 0.0)
# 根据行业查询 行业修正系数与ROE
industry_obj = await Industry.filter(name=data.industry).first()
# roe_score = industry_obj.roe
fix_num_score = industry_obj.fix_num
try:
industry_obj = await asyncio.wait_for(Industry.filter(name=data.industry).first(), timeout=2.0)
except Exception as e:
logger.warning("valuation.industry_fetch_timeout industry={} err={}", data.industry, repr(e))
fix_num_score = getattr(industry_obj, 'fix_num', 0.0) or 0.0
# 根据行业查询 政策匹配度
policy_obj = await Policy.filter(name=data.industry).first()
policy_match_score = policy_obj.score
try:
policy_obj = await asyncio.wait_for(Policy.filter(name=data.industry).first(), timeout=2.0)
except Exception as e:
logger.warning("valuation.policy_fetch_timeout industry={} err={}", data.industry, repr(e))
policy_match_score = getattr(policy_obj, 'score', 0.0) or 0.0
# 提取 经济价值B1 计算参数
input_data_by_b1 = _extract_calculation_params_b1(data)
input_data_by_b1 = await _extract_calculation_params_b1(data)
# ESG关联价值 ESG分 (0-10分)
input_data_by_b1["esg_score"] = esg_score
# 行业修正系数I
@ -67,40 +78,67 @@ async def calculate_valuation(
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_dict = patent_data if isinstance(patent_data, dict) else {}
inner_data = patent_dict.get("data", {}) if isinstance(patent_dict.get("data", {}), dict) else {}
data_list = inner_data.get("dataList", [])
data_list = data_list if isinstance(data_list, list) else []
# 验证 专利剩余年限
# TODO 无法验证 专利剩余保护期>10年(10分)，5-10年(7分)，<5年(3分)；
# 查询专利是否存在
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
# 发展潜力D相关参数 专利数量
# 查询匹配申请号的记录集合
matched = [item for item in data_list if isinstance(item, dict) and item.get("SQH") == getattr(data, 'patent_application_no', None)]
if matched:
patent_count = calculate_patent_usage_score(len(matched))
input_data_by_b1["patent_count"] = float(patent_count)
else:
input_data_by_b1["patent_count"] = 0
input_data_by_b1["patent_count"] = 0.0
except Exception as e:
input_data_by_b1["patent_count"] = 0
logger.warning("valuation.patent_api_error err={}", repr(e))
input_data_by_b1["patent_count"] = 0.0
# 提取 文化价值B2 计算参数
input_data_by_b2 = _extract_calculation_params_b2(data)
input_data_by_b2 = await _extract_calculation_params_b2(data)
# 提取 风险调整系数B3 计算参数
input_data_by_b3 = _extract_calculation_params_b3(data)
input_data_by_b3 = await _extract_calculation_params_b3(data)
# 提取 市场估值C 参数
input_data_by_c = _extract_calculation_params_c(data)
input_data_by_c = await _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)
input_data = {
# 模型估值B 相关参数
"model_data": {
# 经济价值B1 参数
"economic_data": input_data_by_b1,
# 文化价值B2 参数
"cultural_data": input_data_by_b2,
# 风险调整参数 B3
"risky_data": input_data_by_b3,
},
# 市场估值C 参数
"market_data": input_data_by_c,
}
calculator = FinalValueACalculator()
# 计算最终估值A（统一计算）
calculation_result = calculator.calculate_complete_final_value_a(input_data)
# 结构化日志：关键分值
try:
duration_ms = int((time.monotonic() - start_ts) * 1000)
logger.info(
"valuation.calc_done user_id={} duration_ms={} model_value_b={} market_value_c={} final_value_ab={}",
user_id,
duration_ms,
calculation_result.get('model_value_b'),
calculation_result.get('market_value_c'),
calculation_result.get('final_value_ab'),
)
except Exception:
pass
# 创建估值评估记录
result = await user_valuation_controller.create_valuation(
user_id=user_id,
@ -111,13 +149,23 @@ async def calculate_valuation(
result_dict = json.loads(result.model_dump_json())
result_dict['calculation_result'] = calculation_result
result_dict['calculation_input'] = {
'model_data': {
'economic_data': list(input_data.get('model_data', {}).get('economic_data', {}).keys()),
'cultural_data': list(input_data.get('model_data', {}).get('cultural_data', {}).keys()),
'risky_data': list(input_data.get('model_data', {}).get('risky_data', {}).keys()),
},
'market_data': list(input_data.get('market_data', {}).keys()),
}
return Success(data=result_dict, msg="估值计算完成")
except Exception as e:
logger.error("valuation.calc_failed user_id={} err={}", user_id, repr(e))
raise HTTPException(status_code=400, detail=f"计算失败: {str(e)}")
def _extract_calculation_params_b1(data: UserValuationCreate) -> Dict[str, Any]:
async def _extract_calculation_params_b1(data: UserValuationCreate) -> Dict[str, Any]:
"""
从用户提交的数据中提取计算所需的参数
@ -182,13 +230,13 @@ def _extract_calculation_params_b1(data: UserValuationCreate) -> Dict[str, Any]:
'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,
# 这些社交数据暂未来源，置为0，后续接入API再填充
'likes': 0,
'comments': 0,
'shares': 0,
# followers 非当前计算用键，先移除避免干扰
'click_count': int(data.click_count) or 100,
'view_count': int(data.link_views) or 100,
# click_count 与 view_count 目前未参与计算，先移除
# 政策乘数B13相关参数
'implementation_stage': implementation_stage,
@ -196,7 +244,7 @@ def _extract_calculation_params_b1(data: UserValuationCreate) -> Dict[str, Any]:
}
# 获取 文化价值B2 相关参数
def _extract_calculation_params_b2(data: UserValuationCreate) -> Dict[str, Any]:
async def _extract_calculation_params_b2(data: UserValuationCreate) -> Dict[str, Any]:
"""
argrg:
data: 用户提交的估值评估数据
@ -208,18 +256,18 @@ def _extract_calculation_params_b2(data: UserValuationCreate) -> Dict[str, Any]:
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未知")
douyin_views = 0
kuaishou_views= 0
bilibili_views= 0
# 跨界合作深度 品牌联名0.3，科技载体0.5，国家外交礼品1.0
cross_border_depth = float(data.cooperation_depth)
# 纹样基因值B22相关参数
historical_inheritance = "历史传承度HI信息未找到"
# 用户上传资产相关的纹样图片，系数识别处理，得出结构复杂度值 TODO 一下两项未实现配置 识别图片功能
structure_complexity = "结构复杂度SC"
normalized_entropy = "归一化信息熵H"
# 以下三项需由后续模型/服务计算；此处提供默认可计算占位
historical_inheritance = 0.8
structure_complexity = 0.75
normalized_entropy = 0.85
return {
"inheritor_level_coefficient": inheritor_level_coefficient,
"offline_sessions": offline_sessions,
@ -233,7 +281,7 @@ def _extract_calculation_params_b2(data: UserValuationCreate) -> Dict[str, Any]:
}
# 获取 文化价值B2 相关参数
def _extract_calculation_params_b3(data: UserValuationCreate) -> Dict[str, Any]:
async 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获取)
@ -247,10 +295,31 @@ def _extract_calculation_params_b3(data: UserValuationCreate) -> Dict[str, Any]:
# 获取 市场估值C 相关参数
def _extract_calculation_params_c(data: UserValuationCreate) -> Dict[str, Any]:
async def _extract_calculation_params_c(data: UserValuationCreate) -> Dict[str, Any]:
# 市场竞价C1 TODO 暂无
# transaction_data: 交易数据字典(API获取)
# manual_bids: 手动收集的竞价列表(用户填写)
# expert_valuations: 专家估值列表(系统配置)
transaction_data: Dict = None
manual_bids: List[float] = None
expert_valuations: List[float] = None
# 浏览热度分 TODO 需要先确定平台信息
daily_browse_volume = 500.0 # 近7日日均浏览量(默认占位)
collection_count = 50 # 收藏数(默认占位)
# 稀缺性乘数C3 发行量
circulation = data.circulation\
circulation = data.circulation or '限量'
# 时效性衰减C4 参数 用户选择距离最近一次市场活动（交易、报价、评估）的相距时间
last_market_activity = data.last_market_activity
recent_market_activity = data.last_market_activity or '2024-01-15'
return {
# 计算市场竞价C1
# C1 的实现接受 transaction_data={'weighted_average_price': x}
"weighted_average_price": transaction_data,
"manual_bids": manual_bids, # 手动收集的竞价列表 (用户填写)
"expert_valuations": expert_valuations, # 专家估值列表 (系统配置)
# 计算热度系数C2
"daily_browse_volume": daily_browse_volume, # 近7日日均浏览量 (API获取)
"collection_count": collection_count, # 收藏数
"issuance_level": circulation, # 默认 限量发行 计算稀缺性乘数C3
"recent_market_activity": recent_market_activity, # 默认 '近一月' 计算市场估值C
}

50
app/utils/calculation_engine/__init__.py
View File

@ -1,3 +1,53 @@
'''
这是非物质文化遗产IP知识产权评估系统的核心计算引擎包。
'''
from app.utils.calculation_engine.economic_value_b1 import EconomicValueB1Calculator
from app.utils.calculation_engine.economic_value_b1.sub_formulas import (
BasicValueB11Calculator,
TrafficFactorB12Calculator,
PolicyMultiplierB13Calculator
)
from app.utils.calculation_engine.cultural_value_b2 import CulturalValueB2Calculator
from app.utils.calculation_engine.cultural_value_b2.sub_formulas import (
LivingHeritageB21Calculator,
PatternGeneB22Calculator
)
from app.utils.calculation_engine.risk_adjustment_b3 import RiskAdjustmentB3Calculator
from app.utils.calculation_engine.market_value_c import MarketValueCCalculator
from app.utils.calculation_engine.market_value_c.sub_formulas import (
MarketBiddingC1Calculator,
HeatCoefficientC2Calculator,
ScarcityMultiplierC3Calculator,
TemporalDecayC4Calculator
)
from app.utils.calculation_engine.final_value_ab import FinalValueACalculator
__version__ = "1.0.0"
__author__ = "Assessment Team"
__all__ = [
# 经济价值B1模块
"EconomicValueB1Calculator",
"BasicValueB11Calculator",
"TrafficFactorB12Calculator",
"PolicyMultiplierB13Calculator",
# 文化价值B2模块
"CulturalValueB2Calculator",
"LivingHeritageB21Calculator",
"PatternGeneB22Calculator",
# 风险调整系数B3模块
"RiskAdjustmentB3Calculator",
# 市场估值C模块
"MarketValueCCalculator",
"MarketBiddingC1Calculator",
"HeatCoefficientC2Calculator",
"ScarcityMultiplierC3Calculator",
"TemporalDecayC4Calculator",
# 最终估值A模块
"FinalValueACalculator"
]

1
app/utils/calculation_engine/cultural_value_b2/sub_formulas/living_heritage_b21.py
View File

@ -6,7 +6,6 @@
"""
from typing import List

18
app/utils/calculation_engine/final_value_ab/__init__.py Normal file
View File

@ -0,0 +1,18 @@
"""
最终估值A计算包
最终估值A = 模型估值B × 0.7 + 市场估值C × 0.3
```
"""
from .final_value_a import FinalValueACalculator
__all__ = [
"FinalValueACalculator"
]

166
app/utils/calculation_engine/final_value_ab/final_value_a.py Normal file
View File

@ -0,0 +1,166 @@
"""
最终估值A计算模块
最终估值A = 模型估值B × 0.7 + 市场估值C × 0.3
"""
from typing import Dict
import os
import sys
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(current_dir, os.pardir, os.pardir, os.pardir, os.pardir))
if project_root not in sys.path:
sys.path.append(project_root)
try:
# 包内相对导入
from .model_value_b import ModelValueBCalculator
from ..market_value_c import MarketValueCCalculator
except ImportError:
# 直接运行时的绝对导入
from app.utils.calculation_engine.final_value_ab.model_value_b import ModelValueBCalculator
from app.utils.calculation_engine.market_value_c import MarketValueCCalculator
class FinalValueACalculator:
"""最终估值A计算器"""
def __init__(self):
"""初始化计算器"""
self.model_value_calculator = ModelValueBCalculator()
self.market_value_calculator = MarketValueCCalculator()
def calculate_final_value_a(self,
model_value_b: float,
market_value_c: float) -> float:
"""
计算最终估值A
最终估值A = 模型估值B × 0.7 + 市场估值C × 0.3
model_value_b: 模型估值B (系统计算)
market_value_c: 市场估值C (系统计算)
float: 最终估值A
"""
final_value = model_value_b * 0.7 + market_value_c * 0.3
return final_value
def calculate_complete_final_value_a(self, input_data: Dict) -> Dict:
"""
计算完整的最终估值A，包含所有子模块
input_data: 输入数据字典，包含所有必要的参数
包含所有中间计算结果和最终结果的字典
"""
# 计算模型估值B
model_result = self.model_value_calculator.calculate_complete_model_value_b(
input_data['model_data']
)
model_value_b = model_result['model_value_b']
# 计算市场估值C
market_result = self.market_value_calculator.calculate_complete_market_value_c(
input_data['market_data']
)
market_value_c = market_result['market_value_c']
# 计算最终估值A
final_value_a = self.calculate_final_value_a(
model_value_b,
market_value_c
)
return {
'model_value_b': model_value_b,
'market_value_c': market_value_c,
'final_value_ab': final_value_a,
}
# 示例使用
if __name__ == "__main__":
# 创建计算器实例
calculator = FinalValueACalculator()
# 示例数据
input_data = {
# 模型估值B相关数据
'model_data': {
'economic_data': {
# 基础价值B11相关参数
'three_year_income': [1000, 2000, 2222],
'patent_score': 1, # 专利分 (0-10)
'popularity_score': 4.0, # 普及地域分 (0-10)
'infringement_score': 1.0, # 侵权分 (0-10)
'innovation_ratio': 600.0, # 创新投入比 (×100)
'esg_score': 10.0, # ESG分 (0-10)
'patent_count':0.8,
'industry_coefficient': 12.0,
# 流量因子B12相关参数
'search_index_s1': 4500.0,
'industry_average_s2': 5000.0,
'likes': 4,
'comments': 5,
'shares': 6,
# 政策乘数B13相关参数
'policy_match_score': 10.0,
'implementation_stage': 10.0,
'funding_support': 10.0
},
'cultural_data': {
# 活态传承系数B21相关参数（教学频次由以下数据计算）
'inheritor_level_coefficient': 10.0,
'cross_border_depth': 7.0,
'offline_sessions': 1,
'douyin_views': 2,
'kuaishou_views': 3,
'bilibili_views': 4,
# 纹样基因值B22相关参数
'structure_complexity': 0.75,
'normalized_entropy': 0.85,
'historical_inheritance': 0.80
},
'risky_data': {
# 风险调整系数B3相关参数
'highest_price': 100.0,
'lowest_price': 95.0,
'lawsuit_status': '无诉讼',
'inheritor_ages': [45, 60, 75]
}
},
# 市场估值C相关数据
'market_data': {
'average_transaction_price': 50000.0,
'manual_bids': [48000.0, 50000.0, 52000.0],
'expert_valuations': [49000.0, 51000.0, 50000.0],
'daily_browse_volume': 500.0,
'collection_count': 50,
'issuance_level': '限量',
'recent_market_activity': '2024-01-15'
}
}
# 计算最终估值A
result = calculator.calculate_complete_final_value_a(input_data)
print("最终估值A计算结果:")
print(f"模型估值B: {result['model_value_b']:.2f}")
print(f"市场估值C: {result['market_value_c']:.2f}")
print(f"最终估值A: {result['final_value_ab']:.2f}")

127
app/utils/calculation_engine/final_value_ab/model_value_b.py Normal file
View File

@ -0,0 +1,127 @@
"""
模型估值B计算模块
模型估值B = （经济价值B1*0.7+文化价值B2*0.3）*风险调整系数B3
"""
from typing import Dict
from app.utils.calculation_engine import RiskAdjustmentB3Calculator
try:
# 相对导入（当作为包使用时）
from ..economic_value_b1.economic_value_b1 import EconomicValueB1Calculator
from ..cultural_value_b2.cultural_value_b2 import CulturalValueB2Calculator
except ImportError:
# 绝对导入（当直接运行时）
from app.utils.calculation_engine.economic_value_b1.economic_value_b1 import EconomicValueB1Calculator
from app.utils.calculation_engine.cultural_value_b2.cultural_value_b2 import CulturalValueB2Calculator
class ModelValueBCalculator:
"""模型估值B计算器"""
def __init__(self):
"""初始化计算器"""
self.economic_value_calculator = EconomicValueB1Calculator()
self.cultural_value_calculator = CulturalValueB2Calculator()
self.risk_adjustment_calculator = RiskAdjustmentB3Calculator()
def calculate_model_value_b(self,
economic_value_b1: float,
cultural_value_b2: float,
risk_value_b3: float) -> float:
"""
模型估值B = （经济价值B1*0.7+文化价值B2*0.3）*风险调整系数B3
Args:
economic_value_b1: 经济价值B1
cultural_value_b2: 文化价值B2
risk_value_b3 : 风险调整系数B3
Returns:
float: 模型估值B
"""
model_value = (economic_value_b1 * 0.7 + cultural_value_b2 * 0.3) * risk_value_b3
return model_value
def calculate_complete_model_value_b(self, input_data: Dict) -> Dict:
"""
计算完整的模型估值B，包含所有子公式
Args:
input_data: 输入数据字典，包含所有必要的参数
Returns:
Dict: 包含所有中间计算结果和最终结果的字典
"""
# 计算经济价值B1
economic_result = self.economic_value_calculator.calculate_complete_economic_value_b1(
input_data['economic_data']
)
economic_value_b1 = economic_result['economic_value_b1']
# 计算文化价值B2
cultural_result = self.cultural_value_calculator.calculate_complete_cultural_value_b2(
input_data['cultural_data']
)
cultural_value_b2 = cultural_result['cultural_value_b2']
risk_value_result = self.risk_adjustment_calculator.calculate_complete_risky_value_b3(
input_data['risky_data']
)
risk_value_b3 = risk_value_result['risk_adjustment_b3']
# 计算模型估值B
model_value_b = self.calculate_model_value_b(
economic_value_b1,
cultural_value_b2,
risk_value_b3
)
return {
'economic_value_b1': economic_value_b1,
'cultural_value_b2': cultural_value_b2,
'risk_value_b3': risk_value_b3,
'model_value_b': model_value_b,
}
# 示例使用
if __name__ == "__main__":
# 创建计算器实例
calculator = ModelValueBCalculator()
# 示例数据
input_data = {
# 经济价值B1相关数据
'economic_data': {
'financial_value': 68.04,
'legal_strength': 7.90,
'development_potential': 6.00,
'industry_coefficient': 0.25,
'search_index_s1': 4500.0,
'industry_average_s2': 5000.0,
'social_media_spread_s3': 1.07,
'policy_compatibility_score': 9.1
},
# 文化价值B2相关数据
'cultural_data': {
'inheritor_level_coefficient': 10.0,
'teaching_frequency': 7.0,
'cross_border_depth': 7.0,
'structure_complexity': 0.75,
'normalized_entropy': 0.85,
'historical_inheritance': 0.80
}
}
# 计算模型估值B
result = calculator.calculate_complete_model_value_b(input_data)
print("模型估值B计算结果:")
print(f"经济价值B1: {result['economic_value_b1']:.2f}")
print(f"文化价值B2: {result['cultural_value_b2']:.4f}")
print(f"模型估值B: {result['model_value_b']:.2f}")

33
app/utils/calculation_engine/market_value_c/market_value_c.py
View File

@ -1,4 +1,12 @@
from typing import Dict, List, Optional
import sys
import os
# 添加当前目录到Python路径，确保能正确导入子模块
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.append(current_dir)
try:
# 相对导入（当作为包使用时）
from .sub_formulas.market_bidding_c1 import MarketBiddingC1Calculator
@ -69,10 +77,10 @@ class MarketValueCCalculator:
# 计算市场竞价C1
market_bidding_c1 = self.market_bidding_calculator.calculate_market_bidding_c1(
transaction_data={'weighted_average_price': input_data.get('average_transaction_price', 50000.0)},
manual_bids=[input_data.get('average_transaction_price', 50000.0)],
expert_valuations=[input_data.get('average_transaction_price', 50000.0)]
manual_bids=input_data.get('manual_bids', [input_data.get('average_transaction_price', 50000.0)]),
expert_valuations=input_data.get('expert_valuations', [input_data.get('average_transaction_price', 50000.0)])
)
# 计算热度系数C2
heat_coefficient_c2 = self.heat_coefficient_calculator.calculate_heat_coefficient_c2(
input_data.get('daily_browse_volume', 500.0),
@ -81,12 +89,12 @@ class MarketValueCCalculator:
# 计算稀缺性乘数C3
scarcity_multiplier_c3 = self.scarcity_multiplier_calculator.calculate_scarcity_multiplier_c3(
input_data.get('issuance_level', '限量发行')
input_data.get('issuance_level', '限量')
)
# 计算时效性衰减C4
temporal_decay_c4 = self.temporal_decay_calculator.calculate_temporal_decay_c4(
input_data.get('recent_market_activity', '近一月')
input_data.get('recent_market_activity', '2024-01-15')
)
# 计算市场估值C
@ -115,21 +123,18 @@ if __name__ == "__main__":
input_data = {
# 市场竞价C1相关参数
'average_transaction_price': 50000.0, # 平均交易价格
'market_activity_coefficient': 0.8, # 市场活跃度系数
'manual_bids': [48000.0, 50000.0, 52000.0], # 手动收集的竞价
'expert_valuations': [49000.0, 51000.0, 50000.0], # 专家估值
# 热度系数C2相关参数
'search_heat': 0.7, # 搜索热度
'social_media_heat': 0.6, # 社交媒体热度
'media_coverage_heat': 0.5, # 媒体报道热度
'daily_browse_volume': 500.0, # 近7日日均浏览量
'collection_count': 50, # 收藏数
# 稀缺性乘数C3相关参数
'issuance_scarcity': 0.8, # 发行量稀缺性
'circulation_scarcity': 0.7, # 流通率稀缺性
'uniqueness_scarcity': 0.9, # 独特性稀缺性
'issuance_level': '限量', # 资产发行等级
# 时效性衰减C4相关参数
'time_decay_coefficient': 0.3, # 时间衰减系数
'market_activity_decay': 0.2 # 市场活跃度衰减
'recent_market_activity': '2024-01-15' # 最近市场活动时间
}
# 计算市场估值C

10
app/utils/calculation_engine/risk_adjustment_b3/sub_formulas/risk_adjustment_b3.py
View File

@ -157,15 +157,15 @@ class RiskAdjustmentB3Calculator:
def calculate_complete_risky_value_b3(self, input_data: Dict) -> Dict:
# 计算各项风险评分
market_risk = calculator.calculate_market_risk(input_data[""], input_data["lowest_price"])
legal_risk = calculator.calculate_legal_risk(input_data["lawsuit_status"])
inheritance_risk = calculator.calculate_inheritance_risk(input_data["inheritor_ages"])
market_risk = self.calculate_market_risk(input_data["highest_price"], input_data["lowest_price"])
legal_risk = self.calculate_legal_risk(input_data["lawsuit_status"])
inheritance_risk = self.calculate_inheritance_risk(input_data["inheritor_ages"])
# 计算风险评分总和R
risk_score_sum = calculator.calculate_risk_score_sum(market_risk, legal_risk, inheritance_risk)
risk_score_sum = self.calculate_risk_score_sum(market_risk, legal_risk, inheritance_risk)
# 计算风险调整系数B3
risk_adjustment_b3 = calculator.calculate_risk_adjustment_b3(risk_score_sum)
risk_adjustment_b3 = self.calculate_risk_adjustment_b3(risk_score_sum)
return {
'risk_score_sum': risk_score_sum,
'risk_adjustment_b3': risk_adjustment_b3