NicheCompass — 热门赛道时间审计 × 小众方向价值评估器
一、实际应用场景描述
你坐在电脑前复盘过去半年:
- 学了三个月大模型微调(所有人都在聊)
- 刷了两百小时行业资讯(怕错过风口)
- 参加了五个热门研讨会(大家都去)
- 结果:简历上多了一行关键词,但心里空空的
与此同时,你偶然帮朋友修了一次老式胶片相机,花了一个周末,没人知道,也没人点赞,但你记得那种"手指沾着机油、齿轮咔哒转动"的感觉——那是你半年里唯一一次觉得"我在做事"。
你问自己两个问题:
1. 我花了多少时间在追热门?
2. 那些没人注意的小众探索,到底给了我什么?
NicheCompass 就是为这两个问题而生的工程化工具——它不告诉你"该选什么赛道",它给你一套审计框架:记录时间去向,量化两类投入的回报,然后让你自己看见——你的注意力到底在喂养谁,又在荒废谁。
二、引入痛点
痛点 本质问题 本工具解法
FOMO(错失恐惧) 害怕错过风口 → 盲目追热门 时间追踪 + 机会成本显性化
从众惯性 大家做什么我就做什么 热门/小众标签强制分类
热门回报虚高 只看曝光,不看实质收获 多维回报量化(技能/连接/满足感)
小众价值隐形 没人讨论 = 没价值 长期复利系数 + 独特性评分
赛道模糊 不知道自己真正适合什么 数据驱动的赛道推荐
决策焦虑 选热门怕卷,选小众怕饿死 风险-回报四象限分析
三、核心逻辑讲解
理论基础(工程化映射)
1. 注意力经济学(Herbert Simon)
- 信息消耗注意力,注意力成为稀缺资源
- 本工具将注意力(时间)作为核心记账单位
2. 差异化战略(Blue Ocean Strategy)
- 热门 = 红海(竞争激烈,利润薄)
- 小众 = 蓝海(竞争少,价值独特)
- 本工具用"竞争强度 × 独特性"评估赛道质量
3. 刻意练习理论(Ericsson)
- 有效进步需要专注、反馈、挑战
- 热门方向往往碎片化、浅层化
- 本工具追踪"深度工作时间"占比
4. 长期主义(Compound Effect)
- 小众方向的回报往往是滞后的、非线性的
- 本工具引入"复利系数"修正短期偏见
核心算法流程
记录时间投入(按活动细分)
↓
自动分类:热门 vs 小众
(基于:讨论热度 / 从业人数 / 资本关注度)
↓
多维回报评估(每次活动后)
技能增长 | 连接质量 | 市场认可 | 个人满足 | 长期价值
↓
时间汇总 + 回报加权
↓
赛道质量评分:
热门赛道:回报总和 ÷ 时间投入 × (1 - 竞争强度)
小众赛道:回报总和 ÷ 时间投入 × 复利系数 × 独特性
↓
风险-回报四象限分析
↓
个人适配度计算(基于:满足感权重 + 技能匹配)
↓
推荐:核心赛道 + 实验性小众 + 保留的热门
关键指标定义
指标 公式 含义
时间投资比 热门时间 ÷ 总时间 你多大程度在随大流
单位时间回报 总回报 ÷ 总时间 效率
赛道质量分 回报加权 ÷ 时间 × 修正系数 赛道性价比
独特性指数 1 - (从业者数 ÷ 总人口) 越稀有越高
个人适配度 满足感权重 × 0.4 + 技能匹配 × 0.3 + 长期价值 × 0.3 适合你的程度
机会成本 热门时间 × (小众单位回报 - 热门单位回报) 追热门的隐性代价
四、代码模块化(六层架构)
目录结构
nichecompass/ ← 28 个 Python 文件 / ~2300 行
├── main.py ← 入口层:Application 容器 + CLI 调度
├── config.py ← 配置层:赛道分类/权重/阈值/菜单
│
├── models/ ← 数据模型层(@dataclass,不可变)
│ ├── enums.py ← 枚举:方向类型/回报维度/风险等级
│ ├── time_entry.py ← TimeEntry 实体(时间记录)
│ ├── activity.py ← Activity 实体(具体活动)
│ ├── return_record.py ← ReturnRecord 实体(回报评估)
│ ├── niche_profile.py ← NicheProfile 实体(赛道画像)
│ └── career_recommendation.py ← CareerRecommendation 实体
│
├── storage/ ← 持久化层(Repository 模式)
│ └── json_store.py ← CRUD + 原子写入 + 自动备份
│
├── services/ ← 业务层(7 个 Service,单一职责)
│ ├── tracking_service.py ← 时间追踪(开始/结束/标签)
│ ├── classification_service.py ← 热门/小众自动分类
│ ├── return_service.py ← 多维回报评估
│ ├── analysis_service.py ← ★ 核心:赛道分析引擎
│ ├── comparison_service.py ← 热门 vs 小众对比
│ ├── recommendation_service.py ← 赛道推荐引擎
│ └── report_service.py ← 综合报告 + 导出
│
├── cognition/ ← 认知干预层
│ ├── fomo_detector.py ← FOMO 行为检测
│ ├── bandwagon_bias.py ← 从众偏差识别
│ └── value_reframer.py ← 价值重构(热门→小众)
│
├── utils/ ← 工具层(纯函数,零业务依赖)
│ ├── cli_helper.py ← 终端交互
│ ├── dashboard.py ← 终端可视化
│ ├── statistics.py ← 加权/回归/聚类
│ ├── date_util.py ← 时间计算
│ ├── text_processor.py ← 关键词/主题提取
│ └── validator.py ← 输入校验
│
├── README.md / USAGE.md / CHEATSHEET.md
└── demo.py ← 一键跑通全流程
各层职责边界
层 能做什么 不能做什么
"main.py" 组装系统、路由菜单 不写业务逻辑
"config.py" 集中管理常量/分类/权重 不含逻辑
"models/" 数据结构、校验、序列化 不读写文件
"storage/" 读写 JSON、备份、查询 不知道数据含义
"services/" 核心业务逻辑 不直接操作文件
"cognition/" FOMO检测、价值重构 不修改数据
"utils/" 纯函数工具 零业务依赖
核心模块代码展示
1️⃣ 数据模型(不可变 + 自校验)
# models/time_entry.py
@dataclass(frozen=True)
class TimeEntry:
"""时间记录实体 —— 不可变,防止篡改历史"""
entry_id: str
activity_name: str
start_time: datetime
end_time: datetime | None = None # None 表示进行中
direction_type: str = "unknown" # mainstream / niche / hybrid
tags: List[str] = field(default_factory=list)
notes: str = ""
def __post_init__(self):
if not self.activity_name.strip():
raise ValueError("活动名称不能为空")
if self.end_time and self.end_time < self.start_time:
raise ValueError("结束时间不能早于开始时间")
@property
def duration_minutes(self) -> int:
if not self.end_time:
return 0
delta = self.end_time - self.start_time
return int(delta.total_seconds() / 60)
@property
def is_mainstream(self) -> bool:
return self.direction_type == "mainstream"
# models/return_record.py
@dataclass(frozen=False) # 允许更新,因为有修正机制
class ReturnRecord:
"""回报评估实体 —— 每次活动后的多维打分"""
record_id: str
entry_id: str
skill_growth: int = 0 # 1-10:技能增长
connection_quality: int = 0 # 1-10:连接质量
market_recognition: int = 0 # 1-10:市场认可
personal_satisfaction: int = 5 # 1-10:个人满足感
long_term_value: int = 5 # 1-10:长期价值
notes: str = ""
def __post_init__(self):
for field_name, value in self.__dict__.items():
if field_name.endswith(('_growth', '_quality', '_recognition',
'_satisfaction', '_value')):
if not (0 <= value <= 10):
raise ValueError(f"{field_name} 必须在 0-10 之间")
def weighted_return(self, weights: Dict[str, float]) -> float:
"""计算加权回报"""
return (
self.skill_growth * weights.get('skill', 0.25) +
self.connection_quality * weights.get('connection', 0.15) +
self.market_recognition * weights.get('market', 0.20) +
self.personal_satisfaction * weights.get('satisfaction', 0.25) +
self.long_term_value * weights.get('long_term', 0.15)
)
设计意图:时间记录不可变(历史事实),回报评估可修正(认知更新);多维打分避免单一指标偏见。
2️⃣ 分类引擎(热门 vs 小众判定)
# services/classification_service.py
class ClassificationService:
"""方向分类引擎 —— 基于多维度判断热门/小众"""
def __init__(self):
# 热门方向关键词(可配置)
self.mainstream_keywords = {
'ai': ['大模型', 'LLM', '微调', 'RAG', 'Agent'],
'web3': ['区块链', 'NFT', 'DeFi', '元宇宙'],
'career': ['大厂', '上岸', '考公', '跳槽'],
'skills': ['Python', 'React', 'SQL', 'Excel'],
}
# 小众方向特征(低讨论度 + 高专业度)
self.niche_indicators = {
'low_discussion': 0.3, # 讨论热度低于30%
'high_expertise': 0.7, # 专业度高于70%
'small_community': 10000, # 从业者少于1万
'long_history': 10, # 存在超过10年
}
def classify_activity(self, activity_name: str,
description: str = "") -> Tuple[str, float]:
"""
返回:(分类结果, 置信度)
置信度 0-1:越高越确定
"""
text = f"{activity_name} {description}".lower()
# 1. 关键词匹配
mainstream_score = 0
for category, keywords in self.mainstream_keywords.items():
for kw in keywords:
if kw.lower() in text:
mainstream_score += 1
# 2. 语义分析(简化版,实际可用 embedding)
niche_score = self._detect_niche_features(text)
# 3. 综合判断
if mainstream_score >= 2:
return "mainstream", min(0.9, mainstream_score * 0.3)
elif niche_score >= 0.6:
return "niche", niche_score
elif mainstream_score >= 1:
return "hybrid", 0.5 # 混合方向
else:
return "unknown", 0.1
def _detect_niche_features(self, text: str) -> float:
"""检测小众特征"""
score = 0.0
# 特征1:包含专业术语
expert_terms = ['算法', '架构', '范式', '拓扑', '熵']
if any(term in text for term in expert_terms):
score += 0.3
# 特征2:提及具体问题而非泛泛而谈
specific_words = ['如何', '为什么', '原理', '实现', '调试']
if any(word in text for word in specific_words):
score += 0.2
# 特征3:长度较长(通常小众话题描述更详细)
if len(text) > 50:
score += 0.1
return min(1.0, score)
设计意图:不是二元分类,而是概率化判断;支持混合方向(现实中大量存在)。
3️⃣ 分析引擎(核心算法)
# services/analysis_service.py
class AnalysisService:
"""赛道分析引擎 —— 计算各类指标"""
def analyze_direction_performance(self,
entries: List[TimeEntry],
returns: List[ReturnRecord]) -> Dict:
"""分析不同方向的表现"""
# 1. 按方向分组
grouped = self._group_by_direction(entries, returns)
results = {}
for direction, data in grouped.items():
if not data['entries']:
continue
# 2. 基础统计
total_time = sum(e.duration_minutes for e in data['entries'])
avg_return = self._avg_return(data['returns'])
total_return = sum(r.weighted_return(CONFIG.RETURN_WEIGHTS)
for r in data['returns'])
# 3. 赛道质量评分
quality_score = self._calculate_niche_quality(
direction, data, total_time, total_return
)
# 4. 个人适配度
fit_score = self._calculate_personal_fit(data['returns'])
results[direction] = {
'total_time_minutes': total_time,
'entry_count': len(data['entries']),
'avg_return': avg_return,
'total_return': total_return,
'quality_score': quality_score,
'personal_fit': fit_score,
'roi': total_return / max(total_time, 1), # 每分钟回报
'compound_factor': self._estimate_compound_factor(data),
}
return results
def _calculate_niche_quality(self, direction: str,
data: Dict,
total_time: int,
total_return: float) -> float:
"""计算赛道质量分"""
if direction == 'mainstream':
# 热门赛道:回报除以竞争强度
competition_penalty = self._estimate_competition(data)
base_score = (total_return / max(total_time, 1)) * 100
return base_score * (1 - competition_penalty)
elif direction == 'niche':
# 小众赛道:回报乘以复利系数和独特性
uniqueness = self._calculate_uniqueness(data)
compound_factor = self._estimate_compound_factor(data)
base_score = (total_return / max(total_time, 1)) * 100
return base_score * compound_factor * uniqueness
else: # hybrid
return (total_return / max(total_time, 1)) * 80 # 中性评分
def _estimate_competition(self, data: Dict) -> float:
"""估算竞争强度(0-1)"""
# 简化版:基于活动频率和热门关键词
frequency = len(data['entries']) / 30 # 每月平均次数
mainstream_kw_ratio = self._count_mainstream_keywords(data) / len(data['entries'])
# 高频 + 高热度关键词 = 高竞争
return min(0.8, frequency * 0.1 + mainstream_kw_ratio * 0.7)
def _calculate_uniqueness(self, data: Dict) -> float:
"""计算独特性指数(1-3倍)"""
# 基于:小众关键词密度 + 专业深度 + 社区规模
niche_kw_density = self._count_niche_keywords(data) / len(data['entries'])
depth_avg = sum(r.long_term_value for r in data['returns']) / len(data['returns'])
# 独特性 = 基础值(1.0) + 关键词加成 + 深度加成
return 1.0 + niche_kw_density * 1.0 + (depth_avg - 5) * 0.2
def _estimate_compound_factor(self, data: Dict) -> float:
"""估算复利系数(1-2倍)"""
# 小众方向通常有更高的复利效应
consistency = self._calculate_consistency(data) # 连续性
depth = sum(r.long_term_value for r in data['returns']) / len(data['returns'])
# 持续投入 + 高深度 = 高复利
return 1.0 + consistency * 0.5 + (depth - 5) * 0.1
设计意图:热门和小众用不同公式计算,体现其本质差异;引入复利系数修正短期偏见。
4️⃣ 推荐引擎(赛道决策)
# services/recommendation_service.py
class RecommendationService:
"""赛道推荐引擎 —— 基于数据给出建议"""
def generate_recommendations(self,
analysis_results: Dict,
user_preferences: Dict) -> CareerRecommendation:
"""生成个人化发展建议"""
# 1. 找出最佳赛道
best_niche = self._find_best_niche(analysis_results)
best_mainstream = self._find_best_mainstream(analysis_results)
# 2. 风险评估
risk_assessment = self._assess_risks(analysis_results, best_niche)
# 3. 时间重分配建议
reallocation = self._suggest_time_reallocation(
analysis_results, user_preferences
)
# 4. 生成具体行动
actions = self._generate_action_plan(
best_niche, best_mainstream, reallocation
)
return CareerRecommendation(
primary_niche=best_niche,
secondary_direction=best_mainstream,
risk_level=risk_assessment['level'],
time_reallocation=reallocation,
concrete_actions=actions,
reasoning=self._explain_recommendation(analysis_results, best_niche),
confidence=self._calculate_confidence(analysis_results)
)
def _find_best_niche(self, results: Dict) -> Optional[str]:
"""找出最有潜力的小众方向"""
niche_candidates = {
k: v for k, v in results.items()
if k == 'niche' or 'niche' in k
}
if not niche_candidates:
return None
# 评分 = 质量分 × 个人适配度 × 复利系数
scored = {}
for name, data in niche_candidates.items():
score = (
data['quality_score'] * 0.4 +
data['personal_fit'] * 0.4 +
data['compound_factor'] * 0.2
)
scored[name] = score
return max(scored, key=scored.get)
def _suggest_time_reallocation(self,
results: Dict,
preferences: Dict) -> Dict[str, float]:
"""建议时间重分配(百分比)"""
current = {
'mainstream': results.get('mainstream', {}).get('total_time_minutes', 0),
'niche': results.get('niche', {}).get('total_time_minutes', 0),
'hybrid': results.get('hybrid', {}).get('total_time_minutes', 0),
}
total = sum(current.values())
if total == 0:
return {'mainstream': 50, 'niche': 40, 'hybrid': 10}
current_pct = {k: v / total * 100 for k, v in current.items()}
# 基于 ROI 和个人适配度调整
adjustments = {}
for direction, data in results.items():
if direction in current_pct:
roi_weight = data.get('roi', 1.0)
fit_weight = data.get('personal_fit', 5.0) / 10.0
# 高 ROI + 高适配 = 增加时间
adjustment = (roi_weight - 1.0) * 10 + (fit_weight - 0.5) * 20
adjustments[direction] = adjustment
# 应用调整(确保总和仍为100%)
new_pct = {}
for direction in ['mainstream', 'niche', 'hybrid']:
base = current_pct.get(direction, 0)
adj = adjustments.get(direction, 0)
new_pct[direction] = max(0, base + adj)
# 归一化
total_new = sum(new_pct.values())
return {k: v / total_new * 100 for k, v in new_pct.items()}
5️⃣ FOMO 检测器(认知层)
# cognition/fomo_detector.py
class FOMODetector:
"""FOMO 行为检测 —— 识别非理性跟风"""
def detect_fomo_patterns(self,
entries: List[TimeEntry],
returns: List[ReturnRecord]) -> List[Dict]:
"""检测 FOMO 模式"""
patterns = []
# 模式1:高频低回报的热门活动
mainstream_low_return = self._find_low_return_mainstream(entries, returns)
if mainstream_low_return:
patterns.append({
'type': 'high_frequency_low_return',
'description': f'在{mainstream_low_return}上花费大量时间但回报很低',
'severity': 'high',
'suggestion': '考虑减少此类活动,或改变参与方式'
})
# 模式2:突然转向热门方向
sudden_shifts = self._detect_sudden_shifts(entries)
for shift in sudden_shifts:
patterns.append({
'type': 'sudden_shift_to_trending',
'description': f'突然从{shift["from"]}转向热门方向{shift["to"]}',
'severity': 'medium',
'suggestion': '问自己:是真的感兴趣,还是怕错过?'
})
# 模式3:小众方向浅尝辄止
abandoned_niches = self._find_abandoned_niches(entries, returns)
for niche in abandoned_niches:
patterns.append({
'type': 'abandoned_niche',
'description': f'在小众方向{niche}上投入不足就放弃',
'severity': 'medium',
'suggestion': '小众方向需要更长周期才能见效,考虑延长投入时间'
})
return patterns
def _find_low_return_mainstream(self,
entries: List[TimeEntry],
returns: List[ReturnRecord]) -> Optional[str]:
"""找出高频低回报的热门活动"""
mainstream_entries = [e for e in entries if e.is_mainstream]
if len(mainstream_entries) < 5:
return None
# 按活动名分组
by_activity = {}
for entry in mainstream_entries:
name = entry.activity_name
if name not in by_activity:
by_activity[name] = {'entries': [], 'returns': []}
by_activity[name]['entries'].append(entry)
# 匹配回报
return_map = {r.entry_id: r for r in returns}
for name, data in by_activity.items():
data['returns'] = [
return_map[e.entry_id] for e in data['entries']
if e.entry_id in return_map
]
# 找出高频低回报
for name, data in by_activity.items():
if len(data['entries']) >= 5 and data['returns']:
avg_return = sum(r.weighted_return(CONFIG.RETURN_WEIGHTS)
for r in data['returns']) / len(data['returns'])
total_time = sum(e.duration_minutes for e in data['entries'])
# 高频(>20次)且低回报(<4.0)
if len(data['entries']) > 20 and avg_return < 4.0:
return name
return None
6️⃣ 入口层(依赖注入)
# main.py
class Application:
def __init__(self):
# 6 个 Store
self.entry_store = JSONStore(DATA_DIR/"entries.json", ...)
self.activity_store = JSONStore(DATA_DIR/"activities.json", ...)
self.return_store = JSONStore(DATA_DIR/"returns.json", ...)
self.profile_store = JSONStore(DATA_DIR/"profiles.json", ...)
self.recommendation_store = JSONStore(DATA_DIR/"recommendations.json", ...)
self.log_store = JSONStore(DATA_DIR/"logs.json", ...)
# 认知层(3 个组件)
self.fomo_detector = FOMODetector()
self.bandwagon_detector = BandwagonBiasDetector()
self.reframer = ValueReframer()
# 业务层(7 个 Service)
self.tracking_svc = TrackingService(self.entry_store, self.activity_store)
self.classifier_svc = ClassificationService()
self.return_svc = ReturnService(self.return_store)
self.analysis_svc = AnalysisService()
self.comparison_svc = ComparisonService()
self.recommendation_svc = RecommendationService(
self.recommendation_store, self.profile_store
)
self.report_svc = ReportService(
self.entry_store, self.return_store,
self.recommendation_store, self.log_store
)
# CLI
self.cli = CLIHelper()
项目统计
指标 数值
Python 文件 28 个
总代码量 ~2300 行
架构层数 6 层
外部依赖 0(仅标准库)
方向分类 3 类(热门/小众/混合)
回报维度 5 维(技能/连接/市场/满足/长期)
认知偏差检测 FOMO / 从众 / 短期偏见
数据持久化 本地 JSON + 自动备份
五、README 文件
完整 README.md 已包含在项目中,核心内容:
# NicheCompass
把"我该选什么方向"变成"我的时间到底在喂养什么"。
## 安装
python 3.8+,仅标准库
## 快速开始
python demo.py # 看演示
python main.py # 正式使用
## 核心理念
- 注意力是最稀缺的资源
- 热门 ≠ 适合你
- 小众 ≠ 没价值
- 数据是决策的基础
## 核心功能
- 时间追踪(自动分类热门/小众)
- 多维回报评估(不只看钱)
- 赛道质量分析(ROI + 独特性 + 复利)
- 个人适配度计算
- 风险-回报四象限
- 发展建议生成
## 设计原则
- 不贩卖焦虑
- 不鼓吹小众
- 不神化数据
- 本地优先
六、使用说明(USAGE.md 节选)
如何正确记录:
场景 记录方式
学大模型课程 活动名:"吴恩达大模型课第3章",方向:热门
修胶片相机 活动名:"清洁Minolta X-700快门",方向:小众
参加行业大会 活动名:"AI大会Day1",方向:热门
写技术博客 活动名:"写Rust内存管理文章",方向:混合
回报评估怎么填:
维度 问自己
技能增长 我学到新东西了吗?
连接
利用AI解决实际问题,如果你觉得这个工具好用,欢迎关注长安牧笛!