196 lines
7.5 KiB
Python
196 lines
7.5 KiB
Python
"""团队集体经验库
|
|
|
|
功能：
|
|
1. 学习团队历史评审行为（聚合，非个人）
|
|
2. 提取高频风险模式
|
|
3. 为预审助手提供经验支持
|
|
"""
|
|
|
|
import json
|
|
from typing import Dict, List, Optional
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TeamExperienceStore:
|
|
"""团队集体经验存储"""
|
|
|
|
def __init__(self, data_dir: str = "./xingbang_data/team_experience"):
|
|
self.data_dir = Path(data_dir)
|
|
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
self.experience_file = self.data_dir / "team_patterns.json"
|
|
|
|
# 初始化默认经验模式
|
|
if not self.experience_file.exists():
|
|
self._init_default_patterns()
|
|
|
|
def _init_default_patterns(self):
|
|
"""初始化默认团队经验模式"""
|
|
default_patterns = {
|
|
"risk_patterns": {
|
|
"制造业": [
|
|
{
|
|
"risk_type": "应收账款集中度过高",
|
|
"frequency": 23,
|
|
"typical_threshold": "单一客户占比 > 30%",
|
|
"team_action": "要求补充第二担保人或增加保证金",
|
|
},
|
|
{
|
|
"risk_type": "原材料价格波动风险",
|
|
"frequency": 18,
|
|
"typical_threshold": "大宗商品占成本 > 50%",
|
|
"team_action": "要求提供价格锁定合同或套期保值证明",
|
|
},
|
|
],
|
|
"新能源": [
|
|
{
|
|
"risk_type": "政策依赖度高",
|
|
"frequency": 34,
|
|
"typical_threshold": "补贴收入 > 营收 20%",
|
|
"team_action": "降低授信额度或缩短租赁期限",
|
|
},
|
|
{
|
|
"risk_type": "技术迭代淘汰风险",
|
|
"frequency": 21,
|
|
"typical_threshold": "设备技术落后 2 代以上",
|
|
"team_action": "要求租赁物残值担保或回购承诺",
|
|
},
|
|
],
|
|
"交通运输": [
|
|
{
|
|
"risk_type": "油价波动影响",
|
|
"frequency": 15,
|
|
"typical_threshold": "燃油成本 > 营收 40%",
|
|
"team_action": "要求提供运价调整机制证明",
|
|
},
|
|
],
|
|
},
|
|
"approval_thresholds": {
|
|
"资产负债率": {
|
|
"safe": 0.6,
|
|
"warning": 0.7,
|
|
"critical": 0.8,
|
|
"team_consensus": "≤60% 低风险，60-70% 中风险需补充说明，>70% 高风险需额外担保",
|
|
},
|
|
"流动比率": {
|
|
"safe": 1.5,
|
|
"warning": 1.2,
|
|
"critical": 1.0,
|
|
"team_consensus": "≥1.5 优秀，1.2-1.5 合格，<1.2 需关注短期偿债能力",
|
|
},
|
|
"净利润增长率": {
|
|
"safe": 0.15,
|
|
"warning": 0.05,
|
|
"critical": 0.0,
|
|
"team_consensus": "持续两年负增长通常不予通过",
|
|
},
|
|
},
|
|
"industry_red_flags": {
|
|
"光伏": ["产能严重过剩", "欧美双反调查", "地方政府欠补"],
|
|
"房地产": ["三道红线踩线", "土地储备不足", "回款周期超18个月"],
|
|
"餐饮": ["单店模型未跑通", "加盟商纠纷", "食品安全事故"],
|
|
},
|
|
"common_rejection_reasons": [
|
|
{"reason": "担保方资产不足", "count": 45, "solution": "要求补充第二担保人"},
|
|
{"reason": "征信查询次数过多", "count": 32, "solution": "要求客户说明查询原因"},
|
|
{"reason": "涉诉金额过大", "count": 28, "solution": "等待诉讼结果或要求提供和解协议"},
|
|
{"reason": "租赁物适格性存疑", "count": 25, "solution": "要求第三方评估报告"},
|
|
],
|
|
"metadata": {
|
|
"last_updated": datetime.now().isoformat(),
|
|
"total_cases_analyzed": 156,
|
|
"data_source": "2024-2025 年历史评审案例",
|
|
},
|
|
}
|
|
|
|
with open(self.experience_file, "w", encoding="utf-8") as f:
|
|
json.dump(default_patterns, f, ensure_ascii=False, indent=2)
|
|
|
|
logger.info("团队经验模式初始化完成")
|
|
|
|
def get_risk_patterns(self, industry: str = None) -> List[Dict]:
|
|
"""获取风险模式（可按行业过滤）"""
|
|
with open(self.experience_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
risk_patterns = data.get("risk_patterns", {})
|
|
|
|
if industry:
|
|
# 精确匹配或模糊匹配
|
|
for key in risk_patterns:
|
|
if industry in key or key in industry:
|
|
return risk_patterns[key]
|
|
return []
|
|
else:
|
|
# 返回所有模式
|
|
all_patterns = []
|
|
for patterns in risk_patterns.values():
|
|
all_patterns.extend(patterns)
|
|
return all_patterns
|
|
|
|
def get_approval_thresholds(self) -> Dict:
|
|
"""获取团队共识的审批阈值"""
|
|
with open(self.experience_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
return data.get("approval_thresholds", {})
|
|
|
|
def get_industry_red_flags(self, industry: str) -> List[str]:
|
|
"""获取行业红线（团队总结的高危信号）"""
|
|
with open(self.experience_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
industry_flags = data.get("industry_red_flags", {})
|
|
|
|
# 模糊匹配
|
|
for key in industry_flags:
|
|
if industry in key or key in industry:
|
|
return industry_flags[key]
|
|
|
|
return []
|
|
|
|
def get_common_rejection_reasons(self, top_n: int = 10) -> List[Dict]:
|
|
"""获取高频退回原因及解决方案"""
|
|
with open(self.experience_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
reasons = data.get("common_rejection_reasons", [])
|
|
return reasons[:top_n]
|
|
|
|
def update_pattern(self, pattern_type: str, pattern_data: Dict):
|
|
"""更新经验模式（用于持续学习）"""
|
|
with open(self.experience_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
# 更新指定类型的模式
|
|
if pattern_type in data:
|
|
data[pattern_type].update(pattern_data)
|
|
data["metadata"]["last_updated"] = datetime.now().isoformat()
|
|
|
|
with open(self.experience_file, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
logger.info(f"团队经验模式已更新: {pattern_type}")
|
|
|
|
def get_metadata(self) -> Dict:
|
|
"""获取元数据"""
|
|
with open(self.experience_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
return data.get("metadata", {})
|
|
|
|
|
|
# 全局单例
|
|
_team_experience_store: Optional[TeamExperienceStore] = None
|
|
|
|
|
|
def get_team_experience_store() -> TeamExperienceStore:
|
|
"""获取团队经验存储单例"""
|
|
global _team_experience_store
|
|
if _team_experience_store is None:
|
|
_team_experience_store = TeamExperienceStore()
|
|
return _team_experience_store
|