feat(backend): 6-axis dynamic evaluation system with weight engine and template management

This commit is contained in:
selfrelease
2026-07-19 20:59:31 +08:00
parent f4ddcab2ca
commit de53a252e4
11 changed files with 2145 additions and 1 deletions
+10 -1
View File
@@ -56,13 +56,21 @@ from app.routers.knowledge import router as knowledge_router
from app.routers.data_sources import router as data_sources_router
from app.routers.industry_research import router as industry_research_router
from app.routers.funds import router as funds_router
from app.routers.evaluation import router as evaluation_router
from app.schemas.common import error
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理。"""
# startup
# startup — 初始化预设评价模板
from app.core.database import async_session_factory
from app.services.evaluation_presets import seed_evaluation_templates
async with async_session_factory() as session:
await seed_evaluation_templates(session)
await session.commit()
yield
# shutdown
@@ -160,3 +168,4 @@ app.include_router(knowledge_router, prefix="/api/v1")
app.include_router(data_sources_router, prefix="/api/v1")
app.include_router(industry_research_router, prefix="/api/v1")
app.include_router(funds_router, prefix="/api/v1")
app.include_router(evaluation_router, prefix="/api/v1")
+5
View File
@@ -13,8 +13,10 @@ from app.models.customer_plan import CustomerAcquisitionPlan
from app.models.data_source import DataSource
from app.models.decision_sentinel import DecisionSentinel
from app.models.digital_twin import DigitalTwinModel
from app.models.evaluation_template import EvaluationTemplate
from app.models.exit_prediction import ExitPrediction
from app.models.financial_data import FinancialData
from app.models.fund import CompanyFundLink, Fund
from app.models.health_score import HealthScore
from app.models.hypothesis import Hypothesis
from app.models.inquiry import InquiryList
@@ -50,8 +52,11 @@ __all__ = [
"DataSource",
"DecisionSentinel",
"DigitalTwinModel",
"EvaluationTemplate",
"ExitPrediction",
"FinancialData",
"Fund",
"CompanyFundLink",
"FirmProfile",
"FundProfile",
"HealthScore",
+60
View File
@@ -0,0 +1,60 @@
"""评价模板模型。
6 轴动态评价指标体系的模板配置,支持基金类型、存续期、企业阶段、产业赛道、投资策略、投资人类型的动态组合。
"""
import uuid
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
from app.core.types import JSONBType
class EvaluationTemplate(Base):
"""评价模板 — 6 轴配置单元,定义维度权重和专属指标。"""
__tablename__ = "evaluation_templates"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
tenant_id: Mapped[str] = mapped_column(String(36), ForeignKey("tenants.id"), nullable=False, index=True)
name: Mapped[str] = mapped_column(String(200), nullable=False, comment="模板名称")
# 6 轴参数
fund_type: Mapped[str] = mapped_column(String(50), nullable=False, comment="基金类型:angel/early_vc/growth_vc/pe/cvc/fof/distress/esg")
fund_lifecycle: Mapped[str] = mapped_column(String(50), nullable=False, default="investment", comment="存续期阶段:investment/growth/exit_preparation/liquidation")
company_stage: Mapped[str] = mapped_column(String(50), nullable=False, default="a", comment="企业阶段:seed/a/b/c/pre_ipo")
industry: Mapped[str] = mapped_column(String(50), nullable=False, default="ai", comment="产业赛道:ai/saas/hardware/biotech/consumer/fintech/manufacturing")
strategy: Mapped[str] = mapped_column(String(50), nullable=False, default="growth", comment="投资策略:growth/value/empowerment/turnaround")
investor_type: Mapped[str] = mapped_column(String(50), nullable=False, default="investor", comment="投资人类型:gp/post_invest_lead/investor")
# 权重配置 — {dimension_key: weight},归一化后总和 = 1.0
weights_json: Mapped[dict] = mapped_column(JSONBType, nullable=False, comment="14 维度权重(归一化后)")
# 维度裁剪
enabled_dimensions: Mapped[list] = mapped_column(JSONBType, nullable=False, comment="启用的维度 key 列表")
disabled_dimensions: Mapped[list] = mapped_column(JSONBType, nullable=False, default=list, comment="禁用的维度 key 列表")
# 专属指标 — [{key, label, description, data_source}]
custom_metrics_json: Mapped[dict | None] = mapped_column(JSONBType, nullable=True, comment="赛道专属指标定义")
# 修饰因子
lp_focus_metrics: Mapped[list | None] = mapped_column(JSONBType, nullable=True, comment="LP 附加指标列表")
regional_benchmark: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="地域基准标识:china_mainland/us/sea/europe")
# 元数据
is_default: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, comment="是否为该组合的默认模板")
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, comment="是否启用")
version: Mapped[int] = mapped_column(Integer, nullable=False, default=1, comment="版本号")
created_by: Mapped[str | None] = mapped_column(String(36), nullable=True, comment="创建人")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
)
+95
View File
@@ -0,0 +1,95 @@
"""基金模型。
管理基金类型、存续期、LP 构成等信息,支撑评价指标体系的动态权重计算。
"""
import uuid
from datetime import date, datetime, timezone
from sqlalchemy import Boolean, Date, DateTime, Float, ForeignKey, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
from app.core.types import JSONBType
class Fund(Base):
"""基金信息 — 管理基金类型和存续期。"""
__tablename__ = "funds"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
tenant_id: Mapped[str] = mapped_column(String(36), ForeignKey("tenants.id"), nullable=False, index=True)
name: Mapped[str] = mapped_column(String(200), nullable=False, comment="基金名称")
# 基金类型:angel/early_vc/growth_vc/pe/cvc/fof/distress/esg
fund_type: Mapped[str] = mapped_column(String(50), nullable=False, comment="基金类型")
# 投资策略:growth/value/empowerment/turnaround
strategy: Mapped[str] = mapped_column(String(50), nullable=False, default="growth", comment="投资策略")
# 存续期信息
established_date: Mapped[date | None] = mapped_column(Date, nullable=True, comment="基金成立日")
total_lifespan_months: Mapped[int] = mapped_column(Integer, nullable=False, default=84, comment="总存续期(月)")
investment_period_months: Mapped[int] = mapped_column(Integer, nullable=False, default=48, comment="投资期(月)")
# LP 构成 — {government: 30, market: 50, corporate: 20}
lp_composition_json: Mapped[dict | None] = mapped_column(JSONBType, nullable=True, comment="LP 构成百分比")
# 地域
primary_market: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="主要市场:china_mainland/us/sea/europe")
# 状态
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, comment="是否活跃")
# 元数据
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
)
@property
def current_lifecycle(self) -> str:
"""根据当前日期自动计算基金存续期阶段。
返回:investment / growth / exit_preparation / liquidation
"""
if not self.established_date:
return "investment"
today = date.today()
months_elapsed = (today.year - self.established_date.year) * 12 + (today.month - self.established_date.month)
if months_elapsed < self.investment_period_months:
return "investment"
elif months_elapsed < self.total_lifespan_months - 24:
return "growth"
elif months_elapsed < self.total_lifespan_months - 12:
return "exit_preparation"
else:
return "liquidation"
class CompanyFundLink(Base):
"""企业-基金关联 — 一个企业可能被多支基金投资。"""
__tablename__ = "company_fund_links"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
company_id: Mapped[str] = mapped_column(String(36), ForeignKey("companies.id"), nullable=False, index=True)
fund_id: Mapped[str] = mapped_column(String(36), ForeignKey("funds.id"), nullable=False, index=True)
investment_date: Mapped[date | None] = mapped_column(Date, nullable=True, comment="投资日期")
investment_stage: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="投资时企业阶段")
round: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="轮次")
amount: Mapped[float | None] = mapped_column(Float, nullable=True, comment="投资金额(万元)")
ownership_pct: Mapped[float | None] = mapped_column(Float, nullable=True, comment="持股比例(%")
is_current: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, comment="当前是否持有")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
+11
View File
@@ -13,6 +13,7 @@ from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
from app.core.types import JSONBType
from app.core.types import JSONBType as _JSONB # 兼容别名
class HealthScore(Base):
@@ -47,3 +48,13 @@ class HealthScore(Base):
calculated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
# 评价模板关联(向后兼容:旧数据为 NULL)
template_id: Mapped[str | None] = mapped_column(String(36), ForeignKey("evaluation_templates.id"), nullable=True, comment="使用的评价模板 ID")
fund_type: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="冗余存储基金类型,便于查询")
fund_lifecycle: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="冗余存储存续期阶段")
company_stage: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="冗余存储企业阶段")
industry: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="冗余存储产业赛道")
strategy: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="冗余存储投资策略")
custom_metrics_result: Mapped[dict | None] = mapped_column(JSONBType, nullable=True, comment="专属指标评分结果")
lp_focus_result: Mapped[dict | None] = mapped_column(JSONBType, nullable=True, comment="LP 附加指标结果")
+451
View File
@@ -0,0 +1,451 @@
"""评价模板管理 + 评分计算 API。
提供模板 CRUD、权重计算、评分历史查询等接口。
"""
from typing import Any
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.dependencies import get_current_user
from app.models.evaluation_template import EvaluationTemplate
from app.models.user import User
from app.schemas.common import ApiResponse, error, success
from app.services.evaluation_engine import (
calculate_weighted_score,
compute_weights,
get_dimension_score_key,
)
router = APIRouter(prefix="/evaluation", tags=["evaluation"])
# --- 请求/响应模型 ---
class WeightComputeRequest(BaseModel):
"""权重计算请求。"""
fund_type: str = Field(..., description="基金类型")
fund_lifecycle: str = Field("investment", description="存续期阶段")
company_stage: str = Field("a", description="企业阶段")
industry: str = Field("ai", description="产业赛道")
strategy: str = Field("growth", description="投资策略")
investor_type: str = Field("investor", description="投资人类型")
class TemplateCreateRequest(BaseModel):
"""创建模板请求。"""
name: str = Field(..., description="模板名称")
fund_type: str = Field(..., description="基金类型")
fund_lifecycle: str = Field("investment", description="存续期阶段")
company_stage: str = Field("a", description="企业阶段")
industry: str = Field("ai", description="产业赛道")
strategy: str = Field("growth", description="投资策略")
investor_type: str = Field("investor", description="投资人类型")
weights_json: dict[str, float] | None = Field(None, description="自定义权重(不传则自动计算)")
is_default: bool = Field(False, description="是否为默认模板")
class ScoreCalculateRequest(BaseModel):
"""评分计算请求。"""
company_id: str = Field(..., description="企业 ID")
template_id: str | None = Field(None, description="模板 ID(不传则自动匹配)")
fund_type: str | None = Field(None, description="基金类型(无模板时用于自动匹配)")
fund_lifecycle: str | None = Field(None, description="存续期阶段")
company_stage: str | None = Field(None, description="企业阶段")
industry: str | None = Field(None, description="产业赛道")
strategy: str | None = Field(None, description="投资策略")
structured_data: dict[str, Any] = Field(default_factory=dict, description="月报结构化数据")
# --- API 端点 ---
@router.post("/weights/compute", response_model=ApiResponse[dict])
async def compute_evaluation_weights(
req: WeightComputeRequest,
user: User = Depends(get_current_user),
):
"""根据 6 轴参数实时计算权重(不持久化)。"""
result = compute_weights(
fund_type=req.fund_type,
fund_lifecycle=req.fund_lifecycle,
company_stage=req.company_stage,
industry=req.industry,
strategy=req.strategy,
investor_type=req.investor_type,
)
return success(data=result)
@router.get("/templates", response_model=ApiResponse[list])
async def list_templates(
fund_type: str | None = Query(default=None, description="基金类型筛选"),
fund_lifecycle: str | None = Query(default=None, description="存续期阶段筛选"),
company_stage: str | None = Query(default=None, description="企业阶段筛选"),
industry: str | None = Query(default=None, description="产业赛道筛选"),
strategy: str | None = Query(default=None, description="投资策略筛选"),
is_default: bool | None = Query(default=None, description="仅默认模板"),
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取评价模板列表。"""
query = select(EvaluationTemplate).where(
EvaluationTemplate.tenant_id == user.tenant_id,
EvaluationTemplate.is_active == True, # noqa: E712
)
if fund_type:
query = query.where(EvaluationTemplate.fund_type == fund_type)
if fund_lifecycle:
query = query.where(EvaluationTemplate.fund_lifecycle == fund_lifecycle)
if company_stage:
query = query.where(EvaluationTemplate.company_stage == company_stage)
if industry:
query = query.where(EvaluationTemplate.industry == industry)
if strategy:
query = query.where(EvaluationTemplate.strategy == strategy)
if is_default is not None:
query = query.where(EvaluationTemplate.is_default == is_default)
query = query.order_by(EvaluationTemplate.fund_type, EvaluationTemplate.company_stage)
result = await db.execute(query)
templates = result.scalars().all()
return success(data=[
{
"id": str(t.id),
"name": t.name,
"fund_type": t.fund_type,
"fund_lifecycle": t.fund_lifecycle,
"company_stage": t.company_stage,
"industry": t.industry,
"strategy": t.strategy,
"investor_type": t.investor_type,
"weights": t.weights_json,
"enabled_dimensions": t.enabled_dimensions,
"disabled_dimensions": t.disabled_dimensions,
"custom_metrics": t.custom_metrics_json,
"is_default": t.is_default,
"version": t.version,
}
for t in templates
])
@router.get("/templates/{template_id}", response_model=ApiResponse[dict])
async def get_template(
template_id: str,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取单个评价模板详情。"""
result = await db.execute(
select(EvaluationTemplate).where(EvaluationTemplate.id == template_id)
)
tmpl = result.scalar_one_or_none()
if not tmpl:
return error(code=404, message="模板不存在")
return success(data={
"id": str(tmpl.id),
"name": tmpl.name,
"fund_type": tmpl.fund_type,
"fund_lifecycle": tmpl.fund_lifecycle,
"company_stage": tmpl.company_stage,
"industry": tmpl.industry,
"strategy": tmpl.strategy,
"investor_type": tmpl.investor_type,
"weights": tmpl.weights_json,
"enabled_dimensions": tmpl.enabled_dimensions,
"disabled_dimensions": tmpl.disabled_dimensions,
"custom_metrics": tmpl.custom_metrics_json,
"lp_focus_metrics": tmpl.lp_focus_metrics,
"regional_benchmark": tmpl.regional_benchmark,
"is_default": tmpl.is_default,
"is_active": tmpl.is_active,
"version": tmpl.version,
"created_at": tmpl.created_at.isoformat() if tmpl.created_at else None,
"updated_at": tmpl.updated_at.isoformat() if tmpl.updated_at else None,
})
@router.post("/templates", response_model=ApiResponse[dict])
async def create_template(
req: TemplateCreateRequest,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""创建自定义评价模板。"""
# 如果未提供权重,自动计算
if req.weights_json is None:
result = compute_weights(
fund_type=req.fund_type,
fund_lifecycle=req.fund_lifecycle,
company_stage=req.company_stage,
industry=req.industry,
strategy=req.strategy,
investor_type=req.investor_type,
)
weights_json = result["weights"]
enabled_dims = result["enabled_dimensions"]
disabled_dims = result["disabled_dimensions"]
custom_metrics = {"metrics": result["custom_metrics"]}
else:
weights_json = req.weights_json
# 从权重 key 推导启用/禁用维度
all_dims = list(get_dimension_score_key(k).replace("_score", "") for k in weights_json)
enabled_dims = [k for k, v in weights_json.items() if v > 0]
disabled_dims = [k for k in all_dims if k not in enabled_dims]
custom_metrics = None
tmpl = EvaluationTemplate(
tenant_id=user.tenant_id,
name=req.name,
fund_type=req.fund_type,
fund_lifecycle=req.fund_lifecycle,
company_stage=req.company_stage,
industry=req.industry,
strategy=req.strategy,
investor_type=req.investor_type,
weights_json=weights_json,
enabled_dimensions=enabled_dims,
disabled_dimensions=disabled_dims,
custom_metrics_json=custom_metrics,
is_default=req.is_default,
is_active=True,
version=1,
created_by=user.id,
)
db.add(tmpl)
await db.flush()
return success(data={
"id": str(tmpl.id),
"name": tmpl.name,
"weights": tmpl.weights_json,
})
@router.post("/score", response_model=ApiResponse[dict])
async def calculate_score(
req: ScoreCalculateRequest,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""根据模板和月报数据计算评价评分。"""
from app.models.health_score import HealthScore
from app.services.health_calculator import calculate_health_score
# 获取模板
tmpl: EvaluationTemplate | None = None
if req.template_id:
result = await db.execute(
select(EvaluationTemplate).where(EvaluationTemplate.id == req.template_id)
)
tmpl = result.scalar_one_or_none()
if not tmpl:
# 自动匹配模板
query = select(EvaluationTemplate).where(
EvaluationTemplate.tenant_id == user.tenant_id,
EvaluationTemplate.is_active == True, # noqa: E712
EvaluationTemplate.is_default == True,
)
if req.fund_type:
query = query.where(EvaluationTemplate.fund_type == req.fund_type)
if req.fund_lifecycle:
query = query.where(EvaluationTemplate.fund_lifecycle == req.fund_lifecycle)
if req.company_stage:
query = query.where(EvaluationTemplate.company_stage == req.company_stage)
if req.industry:
query = query.where(EvaluationTemplate.industry == req.industry)
result = await db.execute(query.limit(1))
tmpl = result.scalar_one_or_none()
# 计算各维度评分(复用现有 health_calculator
dimension_scores = calculate_health_score(req.structured_data)
# 如果有模板,使用模板权重计算加权总分
if tmpl:
weights = tmpl.weights_json
# 将维度评分转换为权重 key 格式
dim_scores_by_weight_key: dict[str, float] = {}
for weight_key in weights:
score_key = get_dimension_score_key(weight_key)
dim_scores_by_weight_key[weight_key] = dimension_scores.get(score_key, 0.0)
total_score = calculate_weighted_score(dim_scores_by_weight_key, weights)
# 保存评分记录
score_record = HealthScore(
company_id=req.company_id,
total_score=total_score,
financial_score=dimension_scores.get("financial_score"),
operational_score=dimension_scores.get("operational_score"),
ai_commercial_score=dimension_scores.get("ai_commercial_score"),
ai_cost_score=dimension_scores.get("ai_cost_score"),
org_talent_score=dimension_scores.get("org_talent_score"),
product_tech_score=dimension_scores.get("product_tech_score"),
market_compete_score=dimension_scores.get("market_compete_score"),
governance_score=dimension_scores.get("governance_score"),
financing_score=dimension_scores.get("financing_score"),
synergy_score=dimension_scores.get("synergy_score"),
ai_model_product_score=dimension_scores.get("ai_model_product_score"),
data_compliance_score=dimension_scores.get("data_compliance_score"),
team_tech_score=dimension_scores.get("team_tech_score"),
customer_success_score=dimension_scores.get("customer_success_score"),
template_id=tmpl.id,
fund_type=tmpl.fund_type,
fund_lifecycle=tmpl.fund_lifecycle,
company_stage=tmpl.company_stage,
industry=tmpl.industry,
strategy=tmpl.strategy,
evidence_json={"template_name": tmpl.name, "weights": weights},
)
db.add(score_record)
await db.flush()
return success(data={
"score_id": str(score_record.id),
"total_score": total_score,
"dimension_scores": {k: v for k, v in dimension_scores.items() if k != "total_score"},
"template": {
"id": str(tmpl.id),
"name": tmpl.name,
"weights": weights,
},
})
else:
# 无模板,使用默认计算
total_score = dimension_scores.get("total_score", 0.0)
return success(data={
"total_score": total_score,
"dimension_scores": {k: v for k, v in dimension_scores.items() if k != "total_score"},
"template": None,
})
@router.get("/scores", response_model=ApiResponse[list])
async def list_scores(
company_id: str | None = Query(default=None, description="企业 ID 筛选"),
template_id: str | None = Query(default=None, description="模板 ID 筛选"),
limit: int = Query(default=20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取评价评分历史。"""
from app.models.health_score import HealthScore
from app.models.company import Company
query = (
select(HealthScore)
.join(Company, HealthScore.company_id == Company.id)
.where(Company.tenant_id == user.tenant_id)
)
if company_id:
query = query.where(HealthScore.company_id == company_id)
if template_id:
query = query.where(HealthScore.template_id == template_id)
query = query.order_by(HealthScore.calculated_at.desc()).limit(limit)
result = await db.execute(query)
scores = result.scalars().all()
return success(data=[
{
"id": str(s.id),
"company_id": s.company_id,
"total_score": s.total_score,
"financial_score": s.financial_score,
"operational_score": s.operational_score,
"ai_commercial_score": s.ai_commercial_score,
"ai_cost_score": s.ai_cost_score,
"org_talent_score": s.org_talent_score,
"product_tech_score": s.product_tech_score,
"market_compete_score": s.market_compete_score,
"governance_score": s.governance_score,
"financing_score": s.financing_score,
"synergy_score": s.synergy_score,
"ai_model_product_score": s.ai_model_product_score,
"data_compliance_score": s.data_compliance_score,
"team_tech_score": s.team_tech_score,
"customer_success_score": s.customer_success_score,
"trend": s.trend,
"template_id": s.template_id,
"fund_type": s.fund_type,
"fund_lifecycle": s.fund_lifecycle,
"company_stage": s.company_stage,
"industry": s.industry,
"strategy": s.strategy,
"calculated_at": s.calculated_at.isoformat() if s.calculated_at else None,
}
for s in scores
])
# --- 基金管理端点 ---
@router.get("/funds", response_model=ApiResponse[list])
async def list_funds(
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""获取当前租户的基金列表。"""
from app.models.fund import Fund
result = await db.execute(
select(Fund)
.where(Fund.tenant_id == user.tenant_id, Fund.is_active == True) # noqa: E712
.order_by(Fund.established_date.desc())
)
funds = result.scalars().all()
return success(data=[
{
"id": str(f.id),
"name": f.name,
"fund_type": f.fund_type,
"strategy": f.strategy,
"established_date": f.established_date.isoformat() if f.established_date else None,
"total_lifespan_months": f.total_lifespan_months,
"investment_period_months": f.investment_period_months,
"current_lifecycle": f.current_lifecycle,
"lp_composition": f.lp_composition_json,
"primary_market": f.primary_market,
}
for f in funds
])
@router.post("/funds", response_model=ApiResponse[dict])
async def create_fund(
req: dict[str, Any],
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
"""创建基金。"""
from app.models.fund import Fund
from datetime import date
fund = Fund(
tenant_id=user.tenant_id,
name=req.get("name", ""),
fund_type=req.get("fund_type", "early_vc"),
strategy=req.get("strategy", "growth"),
established_date=date.fromisoformat(req["established_date"]) if req.get("established_date") else None,
total_lifespan_months=req.get("total_lifespan_months", 84),
investment_period_months=req.get("investment_period_months", 48),
lp_composition_json=req.get("lp_composition"),
primary_market=req.get("primary_market"),
)
db.add(fund)
await db.flush()
return success(data={
"id": str(fund.id),
"name": fund.name,
"current_lifecycle": fund.current_lifecycle,
})
+313
View File
@@ -0,0 +1,313 @@
"""评价权重计算引擎。
6 轴动态权重合并 + 归一化:
1. 基金类型 × 存续期 → 基础权重模板(32 个预设)
2. 企业阶段 → 阶段系数调整(5 个预设)
3. 产业赛道 → 维度裁剪 + 专属指标注入(6 个预设)
4. 投资策略 → ±5% 微调(4 个预设)
5. 归一化 — 裁剪后剩余维度权重自动归一化到 100%
6. 修饰因子叠加 — LP 附加指标 + 地域基准校准
"""
import logging
from typing import Any
logger = logging.getLogger(__name__)
# --- 基础 14 维度权重(来自 health_calculator.py ---
BASE_WEIGHTS: dict[str, float] = {
"financial": 0.15,
"operational": 0.10,
"ai_commercial": 0.10,
"ai_cost": 0.05,
"org_talent": 0.10,
"product_tech": 0.10,
"market_compete": 0.10,
"governance": 0.05,
"financing": 0.05,
"synergy": 0.05,
"ai_model_product": 0.05,
"data_compliance": 0.05,
"team_tech": 0.03,
"customer_success": 0.07,
}
# 维度 key 映射:权重 key → 评分字段 key
DIMENSION_KEY_MAP: dict[str, str] = {
"financial": "financial_score",
"operational": "operational_score",
"ai_commercial": "ai_commercial_score",
"ai_cost": "ai_cost_score",
"org_talent": "org_talent_score",
"product_tech": "product_tech_score",
"market_compete": "market_compete_score",
"governance": "governance_score",
"financing": "financing_score",
"synergy": "synergy_score",
"ai_model_product": "ai_model_product_score",
"data_compliance": "data_compliance_score",
"team_tech": "team_tech_score",
"customer_success": "customer_success_score",
}
# --- 轴 5:基金类型 × 存续期基础权重系数(32 个预设) ---
FUND_LIFECYCLE_MULTIPLIERS: dict[tuple[str, str], dict[str, float]] = {
("angel", "investment"): {"financial": 0.5, "product_tech": 1.8, "org_talent": 1.5, "market": 1.2},
("angel", "growth"): {"financial": 0.7, "product_tech": 1.5, "org_talent": 1.3, "market": 1.2},
("angel", "exit_preparation"): {"financial": 1.0, "product_tech": 1.2, "org_talent": 1.0, "financing": 1.5},
("angel", "liquidation"): {"financial": 1.5, "financing": 2.0, "product_tech": 0.8},
("early_vc", "investment"): {"financial": 0.7, "product_tech": 1.5, "market": 1.2, "customer_success": 0.8},
("early_vc", "growth"): {"financial": 0.9, "product_tech": 1.3, "market": 1.2, "customer_success": 1.0},
("early_vc", "exit_preparation"): {"financial": 1.3, "market": 1.0, "customer_success": 1.2, "financing": 1.5},
("early_vc", "liquidation"): {"financial": 1.8, "financing": 2.0, "product_tech": 0.6},
("growth_vc", "investment"): {"financial": 1.0, "market": 1.3, "customer_success": 1.2},
("growth_vc", "growth"): {"financial": 1.2, "market": 1.2, "customer_success": 1.3},
("growth_vc", "exit_preparation"): {"financial": 1.5, "market": 1.0, "customer_success": 1.2, "financing": 1.5},
("growth_vc", "liquidation"): {"financial": 2.0, "financing": 2.0, "market": 0.8},
("pe", "investment"): {"financial": 1.8, "governance": 1.5, "customer_success": 1.3, "product_tech": 0.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("pe", "growth"): {"financial": 2.0, "governance": 1.5, "customer_success": 1.3, "product_tech": 0.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("pe", "exit_preparation"): {"financial": 2.5, "governance": 1.8, "financing": 1.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("pe", "liquidation"): {"financial": 3.0, "financing": 2.0, "governance": 1.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("cvc", "investment"): {"synergy": 2.0, "market": 1.3, "product_tech": 1.2, "financial": 0.7},
("cvc", "growth"): {"synergy": 1.8, "market": 1.2, "product_tech": 1.2, "financial": 0.8},
("cvc", "exit_preparation"): {"synergy": 1.5, "market": 1.0, "financial": 1.2, "financing": 1.3},
("cvc", "liquidation"): {"financial": 1.5, "financing": 1.5, "synergy": 1.0},
("distress", "investment"): {"financial": 2.5, "governance": 1.5, "product_tech": 0.5, "market": 0.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("distress", "growth"): {"financial": 2.5, "governance": 1.5, "product_tech": 0.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("distress", "exit_preparation"): {"financial": 3.0, "governance": 1.5, "financing": 1.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("distress", "liquidation"): {"financial": 3.0, "financing": 2.0, "governance": 1.5, "ai_commercial": 0.3, "ai_cost": 0.3, "ai_model_product": 0.3},
("esg", "investment"): {"governance": 1.8, "data_compliance": 1.5, "product_tech": 1.2, "financial": 0.8},
("esg", "growth"): {"governance": 1.5, "data_compliance": 1.3, "customer_success": 1.2},
("esg", "exit_preparation"): {"governance": 1.8, "financial": 1.3, "financing": 1.3},
("esg", "liquidation"): {"financial": 1.5, "governance": 1.5, "financing": 1.5},
# FOF 不直接评价单企业,使用默认权重
("fof", "investment"): {},
("fof", "growth"): {},
("fof", "exit_preparation"): {},
("fof", "liquidation"): {},
}
# --- 轴 2:企业阶段系数(5 个预设) ---
STAGE_MULTIPLIERS: dict[str, dict[str, float]] = {
"seed": {"financial": 0.6, "product_tech": 1.8, "org_talent": 1.5, "market_compete": 0.5, "governance": 0.6, "customer_success": 0.5},
"a": {"financial": 0.8, "product_tech": 1.5, "org_talent": 1.2, "market_compete": 1.0, "governance": 0.7, "customer_success": 0.8},
"b": {"financial": 1.2, "market_compete": 1.3, "customer_success": 1.3, "product_tech": 1.0, "governance": 1.0},
"c": {"financial": 1.5, "market_compete": 1.3, "governance": 1.3, "customer_success": 1.3, "product_tech": 0.8},
"pre_ipo": {"financial": 1.5, "governance": 1.8, "customer_success": 1.3, "market_compete": 1.0, "product_tech": 0.8},
}
# --- 轴 3:产业赛道配置(6 个预设) ---
INDUSTRY_CONFIGS: dict[str, dict[str, Any]] = {
"ai": {
"enabled": list(DIMENSION_KEY_MAP.keys()),
"disabled": [],
"custom_metrics": [
{"key": "model_accuracy", "label": "模型精度", "description": "AI 模型准确率/召回率"},
{"key": "inference_cost", "label": "推理成本", "description": "单次推理成本趋势"},
{"key": "api_call_volume", "label": "API 调用量", "description": "月度 API 调用次数"},
{"key": "poc_conversion_rate", "label": "PoC 转化率", "description": "PoC 到付费转化率"},
],
},
"saas": {
"enabled": ["financial", "operational", "ai_commercial", "org_talent", "product_tech",
"market_compete", "governance", "financing", "synergy", "data_compliance",
"team_tech", "customer_success"],
"disabled": ["ai_cost", "ai_model_product"],
"custom_metrics": [
{"key": "arr_growth", "label": "ARR 增长率", "description": "年度经常性收入增长率"},
{"key": "net_revenue_retention", "label": "NRR", "description": "净收入留存率"},
{"key": "cac_payback", "label": "CAC 回收期", "description": "获客成本回收月数"},
{"key": "rule_of_40", "label": "Rule of 40", "description": "增长率 + 利润率"},
],
},
"hardware": {
"enabled": ["financial", "operational", "org_talent", "product_tech", "market_compete",
"governance", "financing", "synergy", "data_compliance",
"team_tech", "customer_success"],
"disabled": ["ai_commercial", "ai_cost", "ai_model_product"],
"custom_metrics": [
{"key": "patent_count", "label": "专利数", "description": "累计授权专利数量"},
{"key": "tape_out_progress", "label": "流片进度", "description": "芯片流片里程碑进展"},
{"key": "yield_rate", "label": "良率", "description": "产品良率"},
{"key": "rd_investment_ratio", "label": "研发投入比", "description": "研发投入占营收比例"},
],
},
"biotech": {
"enabled": ["financial", "operational", "org_talent", "product_tech", "governance",
"financing", "synergy", "data_compliance", "team_tech"],
"disabled": ["ai_commercial", "ai_cost", "ai_model_product", "market_compete", "customer_success"],
"custom_metrics": [
{"key": "clinical_stage", "label": "临床阶段", "description": "当前临床试验阶段"},
{"key": "pipeline_progress", "label": "管线进度", "description": "在研管线推进情况"},
{"key": "regulatory_milestone", "label": "审批节点", "description": "监管审批里程碑"},
{"key": "patent_landscape", "label": "专利布局", "description": "核心专利布局覆盖度"},
],
},
"consumer": {
"enabled": ["financial", "operational", "org_talent", "product_tech", "market_compete",
"governance", "financing", "synergy", "team_tech", "customer_success"],
"disabled": ["ai_commercial", "ai_cost", "ai_model_product", "data_compliance"],
"custom_metrics": [
{"key": "gmv", "label": "GMV", "description": "月度交易总额"},
{"key": "repurchase_rate", "label": "复购率", "description": "客户复购率"},
{"key": "brand_index", "label": "品牌指数", "description": "品牌知名度/美誉度"},
{"key": "channel_coverage", "label": "渠道覆盖率", "description": "销售渠道覆盖广度"},
],
},
"fintech": {
"enabled": ["financial", "operational", "ai_commercial", "org_talent", "product_tech",
"market_compete", "governance", "financing", "synergy", "ai_model_product",
"data_compliance", "team_tech", "customer_success"],
"disabled": ["ai_cost"],
"custom_metrics": [
{"key": "license_progress", "label": "牌照进度", "description": "金融牌照获取进展"},
{"key": "risk_control_score", "label": "风控指标", "description": "风控模型评分"},
{"key": "compliance_events", "label": "合规事件", "description": "合规事件数量"},
{"key": "npl_ratio", "label": "坏账率", "description": "不良贷款率"},
],
},
"manufacturing": {
"enabled": ["financial", "operational", "org_talent", "product_tech", "market_compete",
"governance", "financing", "synergy", "data_compliance",
"team_tech", "customer_success"],
"disabled": ["ai_commercial", "ai_cost", "ai_model_product"],
"custom_metrics": [
{"key": "capacity_utilization", "label": "产能利用率", "description": "实际产能/设计产能"},
{"key": "delivery_cycle", "label": "交付周期", "description": "订单交付周期天数"},
{"key": "supply_chain_stability", "label": "供应链稳定性", "description": "供应链中断风险评分"},
{"key": "rd_investment_ratio", "label": "研发投入比", "description": "研发投入占营收比例"},
],
},
}
# --- 轴 4:投资策略微调(4 个预设,±5%) ---
STRATEGY_ADJUSTMENTS: dict[str, dict[str, float]] = {
"growth": {"market_compete": 5, "product_tech": 5, "customer_success": 5, "financial": -5, "governance": -5},
"value": {"financial": 5, "governance": 5, "customer_success": 5, "market_compete": -5, "product_tech": -5},
"empowerment": {"synergy": 5, "org_talent": 5, "product_tech": 5, "financial": -5, "market_compete": -5},
"turnaround": {"financial": 5, "governance": 5, "market_compete": -5, "product_tech": -5},
}
def compute_weights(
fund_type: str,
fund_lifecycle: str,
company_stage: str,
industry: str,
strategy: str = "growth",
investor_type: str = "investor",
) -> dict[str, Any]:
"""计算 6 轴动态权重。
Args:
fund_type: 基金类型 — angel/early_vc/growth_vc/pe/cvc/fof/distress/esg
fund_lifecycle: 存续期阶段 — investment/growth/exit_preparation/liquidation
company_stage: 企业阶段 — seed/a/b/c/pre_ipo
industry: 产业赛道 — ai/saas/hardware/biotech/consumer/fintech/manufacturing
strategy: 投资策略 — growth/value/empowerment/turnaround
investor_type: 投资人类型 — gp/post_invest_lead/investor(不影响权重)
Returns:
包含 weights、enabled_dimensions、disabled_dimensions、custom_metrics 的字典
"""
# Step 1:复制基础权重
weights = dict(BASE_WEIGHTS)
# Step 2:基金类型 × 存续期系数
fund_key = (fund_type, fund_lifecycle)
fund_multipliers = FUND_LIFECYCLE_MULTIPLIERS.get(fund_key, {})
for dim, multiplier in fund_multipliers.items():
if dim in weights:
weights[dim] *= multiplier
# Step 3:企业阶段系数
stage_multipliers = STAGE_MULTIPLIERS.get(company_stage, {})
for dim, multiplier in stage_multipliers.items():
if dim in weights:
weights[dim] *= multiplier
# Step 4:产业赛道 — 维度裁剪
industry_config = INDUSTRY_CONFIGS.get(industry, INDUSTRY_CONFIGS["ai"])
enabled_dims = industry_config["enabled"]
disabled_dims = industry_config["disabled"]
# 禁用的维度权重置零
for dim in disabled_dims:
if dim in weights:
weights[dim] = 0.0
# Step 5:投资策略微调(±5%,基于百分比点)
strategy_adj = STRATEGY_ADJUSTMENTS.get(strategy, {})
for dim, adjustment in strategy_adj.items():
if dim in weights and weights[dim] > 0:
# 将百分比点转换为权重调整量
weights[dim] += adjustment / 100.0
# 确保非负
for dim in weights:
weights[dim] = max(0.0, weights[dim])
# Step 6:归一化 — 只对启用维度归一化到 1.0
enabled_weights = {dim: weights[dim] for dim in enabled_dims if dim in weights}
total = sum(enabled_weights.values())
if total > 0:
normalized = {dim: w / total for dim, w in enabled_weights.items()}
else:
# 极端情况:所有权重为零,均分
count = len(enabled_dims) if enabled_dims else 1
normalized = {dim: 1.0 / count for dim in enabled_dims}
# 转换为百分比格式(保留 4 位小数)
final_weights = {dim: round(w * 100, 2) for dim, w in normalized.items()}
logger.info(
"权重计算完成: fund_type=%s, lifecycle=%s, stage=%s, industry=%s, strategy=%s%s",
fund_type, fund_lifecycle, company_stage, industry, strategy, final_weights,
)
return {
"weights": final_weights,
"enabled_dimensions": enabled_dims,
"disabled_dimensions": disabled_dims,
"custom_metrics": industry_config.get("custom_metrics", []),
}
def get_dimension_score_key(weight_key: str) -> str:
"""将权重 key 转换为评分字段 key。"""
return DIMENSION_KEY_MAP.get(weight_key, f"{weight_key}_score")
def calculate_weighted_score(
dimension_scores: dict[str, float],
weights: dict[str, float],
) -> float:
"""根据维度评分和权重计算加权总分。
Args:
dimension_scores: 各维度评分(0-100),key 为权重 key(如 financial
weights: 各维度权重(百分比),key 为权重 key
Returns:
加权总分(0-100
"""
total = 0.0
weight_sum = 0.0
for dim, weight in weights.items():
score = dimension_scores.get(dim)
if score is not None and weight > 0:
total += score * weight
weight_sum += weight
if weight_sum > 0:
return round(total / weight_sum, 1)
return 0.0
+154
View File
@@ -0,0 +1,154 @@
"""评价模板预设配置种子数据。
启动时自动初始化 50 个预设配置单元到数据库。
"""
import logging
from typing import Any
from app.services.evaluation_engine import compute_weights
logger = logging.getLogger(__name__)
# 基金类型列表
FUND_TYPES = ["angel", "early_vc", "growth_vc", "pe", "cvc", "fof", "distress", "esg"]
# 存续期阶段列表
FUND_LIFECYCLES = ["investment", "growth", "exit_preparation", "liquidation"]
# 企业阶段列表
COMPANY_STAGES = ["seed", "a", "b", "c", "pre_ipo"]
# 产业赛道列表
INDUSTRIES = ["ai", "saas", "hardware", "biotech", "consumer", "fintech", "manufacturing"]
# 投资策略列表
STRATEGIES = ["growth", "value", "empowerment", "turnaround"]
# 投资人类型列表
INVESTOR_TYPES = ["gp", "post_invest_lead", "investor"]
# 模板名称中文名
FUND_TYPE_LABELS = {
"angel": "天使/种子基金",
"early_vc": "早期VC",
"growth_vc": "成长期VC",
"pe": "PE/并购基金",
"cvc": "产业基金",
"fof": "母基金",
"distress": "困境/特殊机会基金",
"esg": "ESG/影响力基金",
}
LIFECYCLE_LABELS = {
"investment": "投资期",
"growth": "成长期",
"exit_preparation": "退出准备期",
"liquidation": "清算期",
}
STAGE_LABELS = {
"seed": "种子/天使",
"a": "A轮",
"b": "B轮",
"c": "C轮+",
"pre_ipo": "Pre-IPO",
}
INDUSTRY_LABELS = {
"ai": "AI/SaaS",
"saas": "企业服务",
"hardware": "硬科技/芯片",
"biotech": "生物医药",
"consumer": "消费品牌",
"fintech": "金融科技",
"manufacturing": "新能源/先进制造",
}
STRATEGY_LABELS = {
"growth": "成长型",
"value": "价值型",
"empowerment": "投后赋能型",
"turnaround": "困境反转型",
}
def generate_preset_templates() -> list[dict[str, Any]]:
"""生成全部预设模板配置。
策略:为每个 (fund_type, lifecycle, stage, industry) 组合生成默认模板,
strategy 默认使用 growthinvestor_type 默认使用 investor。
用户可以在前端创建自定义策略/投资人类型的模板。
Returns:
模板配置字典列表
"""
templates: list[dict[str, Any]] = []
for fund_type in FUND_TYPES:
for lifecycle in FUND_LIFECYCLES:
for stage in COMPANY_STAGES:
for industry in INDUSTRIES:
# FOF 不生成单企业模板
if fund_type == "fof":
continue
result = compute_weights(
fund_type=fund_type,
fund_lifecycle=lifecycle,
company_stage=stage,
industry=industry,
strategy="growth",
investor_type="investor",
)
name = (
f"{FUND_TYPE_LABELS[fund_type]}-{LIFECYCLE_LABELS[lifecycle]}-"
f"{STAGE_LABELS[stage]}-{INDUSTRY_LABELS[industry]}"
)
templates.append({
"name": name,
"fund_type": fund_type,
"fund_lifecycle": lifecycle,
"company_stage": stage,
"industry": industry,
"strategy": "growth",
"investor_type": "investor",
"weights_json": result["weights"],
"enabled_dimensions": result["enabled_dimensions"],
"disabled_dimensions": result["disabled_dimensions"],
"custom_metrics_json": {"metrics": result["custom_metrics"]},
"is_default": True,
"is_active": True,
"version": 1,
})
logger.info("生成预设模板 %d", len(templates))
return templates
async def seed_evaluation_templates(db_session) -> None:
"""将预设模板写入数据库(仅当表为空时)。
Args:
db_session: 异步数据库会话
"""
from sqlalchemy import select
from app.models.evaluation_template import EvaluationTemplate
# 检查是否已有数据
result = await db_session.execute(select(EvaluationTemplate).limit(1))
existing = result.scalar_one_or_none()
if existing:
logger.info("评价模板表已有数据,跳过种子初始化")
return
templates = generate_preset_templates()
for tmpl_data in templates:
tmpl = EvaluationTemplate(**tmpl_data)
db_session.add(tmpl)
await db_session.flush()
logger.info("预设评价模板已写入数据库: %d", len(templates))