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

增加模型计算日志 根据前端内容修改参数

Browse Source
This commit is contained in:
dubingyan666 2025-10-25 17:41:11 +08:00
parent 739b5a80f3
commit 089d6dd504
6 changed files with 151 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
app/api/v1/app_valuations/app_valuations.py
View File

@ -142,6 +142,7 @@ async def _perform_valuation_calculation(user_id: int, data: UserValuationCreate
"market_data": input_data_by_c,
}
calculator = FinalValueACalculator()
# 计算最终估值A（统一计算）
calculation_result = calculator.calculate_complete_final_value_a(input_data)
@ -346,7 +347,7 @@ async def _extract_calculation_params_b1(data: UserValuationCreate) -> Dict[str,
# 行业均值S2 - 从数据库查询行业数据计算
from app.utils.industry_calculator import calculate_industry_average_s2
industry_average_s2 = await calculate_industry_average_s2(data.industry)
# 社交媒体传播度S3 - TODO 需要使用第三方API,click_count view_count 未找到对应参数
# 社交媒体传播度S3
# likes: 点赞数(API获取)
# comments: 评论数(API获取)
# shares: 转发数(API获取)
@ -434,10 +435,11 @@ async def _extract_calculation_params_b2(data: UserValuationCreate) -> Dict[str,
# 纹样基因值B22相关参数
# 以下三项需由后续模型/服务计算；此处提供默认可计算占位
# 历史传承度HI(用户填写)
historical_inheritance = sum([safe_float(i) for i in data.historical_evidence])
structure_complexity = 0.75
normalized_entropy = 0.85
structure_complexity = 0.75 # TODO 需要依赖图片和模型 结构复杂度SC(系统计算)
normalized_entropy = 0.85 # TODO 需要依赖图片和模型 归一化信息熵H(系统计算)
return {
"inheritor_level_coefficient": inheritor_level_coefficient,
"offline_sessions": offline_sessions,
@ -477,16 +479,17 @@ async def _extract_calculation_params_c(data: UserValuationCreate) -> Dict[str,
# TODO 需要客户确认 三个数值
expert_valuations = None
# 浏览热度分 TODO 需要先确定平台信息
daily_browse_volume = 500.0 # 近7日日均浏览量(默认占位)
collection_count = 50 # 收藏数(默认占位)
daily_browse_volume = 500.0 # 近7日日均浏览量(默认占位) 是否 使用该商品近12个月的链接浏览量？？？
collection_count = 50 # 收藏数(默认占位) 无
# 稀缺性乘数C3 发行量
circulation = data.circulation or '限量'
circulation = data.scarcity_level or '限量'
# 时效性衰减C4 参数 用户选择距离最近一次市场活动（交易、报价、评估）的相距时间
recent_market_activity = data.last_market_activity
recent_market_activity = data.market_activity_time
# 如果为空、None或"0"，设置默认值
if not recent_market_activity or recent_market_activity == "0":
recent_market_activity = '2024-01-15'
# recent_market_activity = '2024-01-15'
recent_market_activity = '近一月'
return {
# 计算市场竞价C1
# C1 的实现接受 transaction_data={'weighted_average_price': x}

8
app/schemas/valuation.py
View File

@ -24,7 +24,7 @@ class ValuationAssessmentBase(BaseModel):
heritage_level: Optional[str] = Field(None, description="非遗等级")
heritage_asset_level: Optional[str] = Field(None, description="非遗资产等级")
patent_application_no: Optional[str] = Field(None, description="非遗资产所用专利的申请号")
patent_remaining_years: Optional[str] = Field(None, description="专利剩余年限")
patent_remaining_years: Optional[str] = Field(None, description="专利剩余年限") # 未使用
historical_evidence: Optional[Dict[str, int]] = Field(None, description="非遗资产历史证明证据及数量")
patent_certificates: Optional[List[str]] = Field(None, description="非遗资产所用专利的证书")
pattern_images: Optional[List[str]] = Field(None, description="非遗纹样图片")
@ -37,7 +37,7 @@ class ValuationAssessmentBase(BaseModel):
cooperation_depth: Optional[str] = Field(None, description="非遗资产跨界合作深度")
collaboration_type: Optional[str] = Field(None, description="非遗资产跨界合作深度")
offline_activities: Optional[str] = Field(None, description="近12个月线下相关宣讲活动次数")
offline_teaching_count: Optional[int] = Field(None, description="近12个月线下相关演讲活动次数")
offline_teaching_count: Optional[int] = Field(None, description="近12个月线下相关演讲活动次数") # 未使用
online_accounts: Optional[List[Any]] = Field(None, description="线上相关宣传账号信息")
platform_accounts: Optional[Dict[str, Dict[str, Union[str, int]]]] = Field(None, description="线上相关宣传账号信息")
@ -51,8 +51,8 @@ class ValuationAssessmentBase(BaseModel):
monthly_transaction: Optional[str] = Field(None, description="月交易额")
monthly_transaction_amount: Optional[str] = Field(None, description="月交易额")
price_fluctuation: Optional[List[Union[str, int, float]]] = Field(None, description="该商品近30天价格波动区间")
price_range: Optional[Dict[str, Union[int, float]]] = Field(None, description="资产商品的价格波动率")
market_price: Optional[Union[int, float]] = Field(None, description="市场价格（单位：万元）")
price_range: Optional[Dict[str, Union[int, float]]] = Field(None, description="资产商品的价格波动率") # 未使用
market_price: Optional[Union[int, float]] = Field(None, description="市场价格（单位：万元）") # 未使用
# 内置API计算字段
infringement_record: Optional[str] = Field(None, description="侵权记录")

144
app/utils/calculation_engine/final_value_ab/final_value_a.py
View File

@ -8,7 +8,7 @@
from typing import Dict
import os
import sys
from app.log.log import logger
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:
@ -46,8 +46,20 @@ class FinalValueACalculator:
float: 最终估值A
"""
logger.info("final_value_a.calculate_final_value_a 开始计算最终估值A: 模型估值B={}万元 市场估值C={}万元",
model_value_b, market_value_c)
final_value = model_value_b * 0.7 + market_value_c * 0.3
# 计算加权值
model_weighted = model_value_b * 0.7
market_weighted = market_value_c * 0.3
logger.info("final_value_a.weighted_values 加权计算: 模型估值B加权值={}万元(权重0.7) 市场估值C加权值={}万元(权重0.3)",
model_weighted, market_weighted)
final_value = model_weighted + market_weighted
logger.info("final_value_a.final_calculation 最终估值A计算: 模型估值B={}万元 市场估值C={}万元 模型加权值={}万元 市场加权值={}万元 最终估值AB={}万元",
model_value_b, market_value_c, model_weighted, market_weighted, final_value)
return final_value
@ -60,23 +72,125 @@ class FinalValueACalculator:
包含所有中间计算结果和最终结果的字典
"""
import time
start_time = time.time()
# 记录输入参数
logger.info("final_value_a.calculation_start input_data_keys={} model_data_keys={} market_data_keys={}",
list(input_data.keys()),
list(input_data.get('model_data', {}).keys()),
list(input_data.get('market_data', {}).keys()))
# 详细记录模型数据参数
model_data = input_data.get('model_data', {})
if 'economic_data' in model_data:
economic_data = model_data['economic_data']
logger.info("final_value_a.economic_data 经济价值B1参数: 近三年机构收益={} 专利分={} 普及地域分={} 侵权分={} 创新投入比={} ESG分={} 专利使用量={} 行业修正系数={}",
economic_data.get('three_year_income'),
economic_data.get('patent_score'),
economic_data.get('popularity_score'),
economic_data.get('infringement_score'),
economic_data.get('innovation_ratio'),
economic_data.get('esg_score'),
economic_data.get('patent_count'),
economic_data.get('industry_coefficient'))
if 'cultural_data' in model_data:
cultural_data = model_data['cultural_data']
logger.info("final_value_a.cultural_data 文化价值B2参数: 传承人等级系数={} 跨境深度={} 线下教学次数={} 抖音浏览量={} 快手浏览量={} 哔哩哔哩浏览量={} 结构复杂度={} 归一化信息熵={} 历史传承度={}",
cultural_data.get('inheritor_level_coefficient'),
cultural_data.get('cross_border_depth'),
cultural_data.get('offline_sessions'),
cultural_data.get('douyin_views'),
cultural_data.get('kuaishou_views'),
cultural_data.get('bilibili_views'),
cultural_data.get('structure_complexity'),
cultural_data.get('normalized_entropy'),
cultural_data.get('historical_inheritance'))
if 'risky_data' in model_data:
risky_data = model_data['risky_data']
logger.info("final_value_a.risky_data 风险调整B3参数: 最高价={} 最低价={} 诉讼状态={} 传承人年龄={}",
risky_data.get('highest_price'),
risky_data.get('lowest_price'),
risky_data.get('lawsuit_status'),
risky_data.get('inheritor_ages'))
# 详细记录市场数据参数
market_data = input_data.get('market_data', {})
logger.info("final_value_a.market_data 市场估值C参数: 平均交易价={} 手动出价={} 专家估值={} 日浏览量={} 收藏数量={} 发行等级={} 最近市场活动={}",
market_data.get('average_transaction_price'),
market_data.get('manual_bids'),
market_data.get('expert_valuations'),
market_data.get('daily_browse_volume'),
market_data.get('collection_count'),
market_data.get('issuance_level'),
market_data.get('recent_market_activity'))
# 计算模型估值B
model_result = self.model_value_calculator.calculate_complete_model_value_b(
input_data['model_data']
)
model_value_b = model_result['model_value_b']
logger.info("final_value_a.calculating_model_value_b 开始计算模型估值B")
model_start_time = time.time()
try:
model_result = self.model_value_calculator.calculate_complete_model_value_b(
input_data['model_data']
)
model_value_b = model_result['model_value_b']
model_duration = time.time() - model_start_time
logger.info("final_value_a.model_value_b_calculated 模型估值B计算完成: 模型估值B={}万元 耗时={}ms 返回字段={}",
model_value_b,
int(model_duration * 1000),
list(model_result.keys()))
except Exception as e:
logger.error("final_value_a.model_value_b_calculation_failed 模型估值B计算失败: 错误={} 输入数据={}", str(e), input_data.get('model_data', {}))
raise
# 计算市场估值C
market_result = self.market_value_calculator.calculate_complete_market_value_c(
input_data['market_data']
)
market_value_c = market_result['market_value_c']
logger.info("final_value_a.calculating_market_value_c 开始计算市场估值C")
market_start_time = time.time()
try:
market_result = self.market_value_calculator.calculate_complete_market_value_c(
input_data['market_data']
)
market_value_c = market_result['market_value_c']
market_duration = time.time() - market_start_time
logger.info("final_value_a.market_value_c_calculated 市场估值C计算完成: 市场估值C={}万元 耗时={}ms 返回字段={}",
market_value_c,
int(market_duration * 1000),
list(market_result.keys()))
except Exception as e:
logger.error("final_value_a.market_value_c_calculation_failed 市场估值C计算失败: 错误={} 输入数据={}", str(e), input_data.get('market_data', {}))
raise
# 计算最终估值A
final_value_a = self.calculate_final_value_a(
model_value_b,
market_value_c
)
logger.info("final_value_a.calculating_final_value_a 开始计算最终估值A: 模型估值B={}万元 市场估值C={}万元",
model_value_b, market_value_c)
try:
final_value_a = self.calculate_final_value_a(
model_value_b,
market_value_c
)
# 记录最终计算结果
total_duration = time.time() - start_time
logger.info("final_value_a.calculation_completed 最终估值A计算完成: 最终估值AB={}万元 模型估值B={}万元 市场估值C={}万元 总耗时={}ms 模型计算耗时={}ms 市场计算耗时={}ms",
final_value_a,
model_value_b,
market_value_c,
int(total_duration * 1000),
int(model_duration * 1000),
int(market_duration * 1000))
except Exception as e:
logger.error("final_value_a.final_value_calculation_failed 最终估值A计算失败: 错误={} 模型估值B={}万元 市场估值C={}万元",
str(e), model_value_b, market_value_c)
raise
return {
'model_value_b': model_value_b,
@ -149,7 +263,7 @@ if __name__ == "__main__":
'daily_browse_volume': 500.0,
'collection_count': 50,
'issuance_level': '限量',
'recent_market_activity': '2024-01-15'
'recent_market_activity': '近一月'
}
}

10
app/utils/calculation_engine/market_value_c/sub_formulas/market_bidding_c1.py
View File

@ -61,15 +61,7 @@ if __name__ == "__main__":
transaction_volume = 500 # 交易量 (API获取)
transaction_frequency = "中频" # 交易频率 (用户填写)
market_liquidity = "中" # 市场流动性 (用户填写)
# 获取平均交易价格
average_transaction_price = calculator.get_average_transaction_price(asset_type, time_period)
# 计算市场活跃度系数
market_activity_coefficient = calculator.calculate_market_activity_coefficient(
transaction_volume, transaction_frequency, market_liquidity
)
# 示例数据
# 优先级1：近3个月交易数据
transaction_data = {"weighted_average_price": 1000.0} # API获取

5
app/utils/calculation_engine/market_value_c/sub_formulas/scarcity_multiplier_c3.py
View File

@ -50,8 +50,11 @@ class ScarcityMultiplierC3Calculator:
"稀有": 0.4,
"流通": 0.1
}
for s_key,s_value in scarcity_scores.items():
if s_key in issuance_level:
return s_value
return scarcity_scores.get(issuance_level, 0.1)
return 0.1
# 示例使用

2
app/utils/calculation_engine/market_value_c/sub_formulas/temporal_decay_c4.py
View File

@ -30,7 +30,7 @@ class TemporalDecayC4Calculator:
"其他": 0.1
}
return timeliness_coefficients.get(self.classify_date_distance(time_elapsed), 0.1)
return timeliness_coefficients.get(time_elapsed, 0.1)
@staticmethod
def classify_date_distance(target_date):
"""