夜雨聆风
我用 AI 助理实现财务自由第一步:Nanobot 24 小时股市监控实战,零成本跑赢 90% 散户
我用 AI 助理实现财务自由第一步：Nanobot 24 小时股市监控实战，零成本跑赢 90% 散户
导读：还在为错过最佳买卖点懊悔？还在熬夜看美股财报？本文揭秘如何用 nanobot 搭建一个全天候股市监控分析系统，自动跟踪持仓、预警异动、生成研报，让你睡觉时也有”私人量化团队”在工作。无需金融背景，普通上班族也能实现的智能投资方案。
【 扫描文末二维码加入星球获取源码 】
一、为什么需要 24 小时 AI 助理？
想象一下这样的场景：
-
• 凌晨 3 点，美联储议息会议刚结束，你的 AI 已分析完声明要点，评估对持仓股票的影响，并发送操作建议到你的微信 -
• 早上 7 点，醒来收到昨晚的美股复盘报告：AAPL 财报超预期但指引保守，TSLA 交付量创新高，自动整理卖方观点 -
• 上午 9:30，A50 期货异动，AI 提前 5 分钟预警，提示关注券商板块，你及时跟进获利 3% -
• 下午 2 点，AI 监测到某持仓股票大宗交易折价超 10%，立即推送警告并附上历史数据对比 -
• 晚上 8 点，自动生成今日市场日报：资金流向、北向动态、技术面分析，比你更懂你的持仓
这不是未来科技，而是 nanobot 现在就能做到的事情。
股民的真实痛点
传统炒股模式：
-
• ❌ 信息滞后：等看到新闻时，主力早已布局完成 -
• ❌ 情绪化交易：追涨杀跌，缺乏纪律性 -
• ❌ 时间不够：上班无法盯盘，错过关键买卖点 -
• ❌ 分析能力弱：看不懂财报，不会技术分析 -
• ❌ 精力分散：同时跟踪多只股票力不从心
而 nanobot 带来的变革：
-
• ✅ 7×24 小时监控：全球市场无休，重大事件秒级响应 -
• ✅ 理性决策：基于数据和规则，克服人性弱点 -
• ✅ 自动化执行：条件单、网格交易、再平衡策略自动运行 -
• ✅ 机构级分析：财报解读、估值建模、舆情监控一键生成 -
• ✅ 个性化跟踪：专注你的持仓圈，深度而非广度
二、Nanobot 是什么？
Nanobot 是一个超轻量级的个人 AI 助理框架，专为智能投资场景优化：
|
|
|
|
|---|---|---|
| 门槛 |
|
|
| 成本 |
|
|
| 灵活性 |
|
|
| 部署 |
|
|
| 学习曲线 |
|
|
技术栈一览
Python >= 3.11 # 现代异步编程
LiteLLM # 统一 LLM 接口（支持 100+ 模型）
Pydantic # 配置验证
WebSockets # 实时通信
Croniter # 定时任务调度
MCP (Model Context Protocol) # 外部工具集成投资者专属特性
-
1. 多市场数据源：Yahoo Finance、Alpha Vantage、聚宽、Tushare 无缝集成 -
2. 实时新闻情感分析：Twitter、Reddit、雪球、财联社舆情监控 -
3. 技术指标计算：MACD、KDJ、布林带等 50+ 指标自动绘制 -
4. 财报自动解读：SEC EDGAR、巨潮资讯 API 对接，关键数据提取 -
5. 预警系统：价格、成交量、涨跌幅、资金流向多维度触发 -
6. 投资组合管理：再平衡、止盈止损、对冲策略自动执行
三、内核原理：如何做到 24 小时股市监控？
3.1 整体架构
┌─────────────────────────────────────────────────────┐
│ Channels Layer │
│ Telegram │ Discord │ WhatsApp │ Feishu │ Slack ... │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Message Bus │
│ 事件驱动的消息队列（发布/订阅模式） │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Agent Loop │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Context │ │ Memory │ │ Tools │ │
│ │ Builder │ │Consolidator│ │Registry │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Heartbeat│ │ Cron │ │SubAgents │ │
│ │ Service │ │ Service │ │ Manager │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Provider Layer │
│ OpenAI │ Claude │ Gemini │ Ollama │ vLLM ... │
└─────────────────────────────────────────────────────┘3.2 核心组件详解
（1）Agent Loop：智能分析引擎
nanobot/agent/loop.py 是整个系统的核心大脑，在投资场景下负责：
class AgentLoop:
"""
代理循环核心处理引擎
投资决策流程：
1. 从消息总线接收市场数据/新闻/用户指令
2. 构建上下文（历史持仓 + 记忆 + 分析工具）
3. 调用 LLM 进行推理分析
4. 执行工具调用（获取实时价格、计算指标等）
5. 发送投资建议到聊天软件
"""
async def _run_agent_loop(self, initial_messages, on_progress=None):
messages = initial_messages
iteration = 0
while iteration < self.max_iterations:
iteration += 1
# 1. 获取分析工具定义
tool_defs = self.tools.get_definitions()
# 2. 调用 LLM 分析市场
response = await self.provider.chat_with_retry(
messages=messages,
tools=tool_defs,
model=self.model,
)
# 3. 处理工具调用
if response.has_tool_calls:
for tool_call in response.tool_calls:
result = await self.tools.execute(
tool_call.name,
tool_call.arguments
)
messages = self.context.add_tool_result(
messages,
tool_call.id,
tool_call.name,
result
)
else:
# 4. 获得最终投资建议
final_content = response.content
break
return final_content, tools_used, messages投资场景的关键设计：
-
• 迭代式深度分析：最多支持 40 次工具调用，层层深入分析（如：价格异动→查新闻→看财报→比历史） -
• 错误恢复机制：数据获取失败时自动重试或切换数据源 -
• 实时进度推送：通过 on_progress回调实时推送分析过程（”正在获取 AAPL 实时价格…” → “计算技术指标…” → “分析完成”） -
• 会话隔离：每个用户的投资组合独立管理，隐私安全
（2）记忆系统：投资知识库
nanobot/agent/memory.py 实现了类人记忆机制，在投资场景下：
class MemoryStore:
"""双层投资记忆存储"""
# 第一层：MEMORY.md - 长期投资事实
# 示例内容：
# - 用户风险偏好：中等风险，最大回撤容忍 15%
# - 持仓股票：AAPL(10%), TSLA(15%), NVDA(20%)
# - 投资策略：价值投资为主，辅以趋势交易
# - 历史决策：2026-02 清仓 META，获利 45%
# 第二层：HISTORY.md - 可搜索的交易日志
# 示例内容：
# [2026-03-18 10:30] 买入 NVDA 理由：AI 芯片需求爆发，Q1 财报预期乐观
# [2026-03-18 14:20] 减仓 TSLA：交付量不及预期，技术面破位记忆整合流程（投资版）：
async def consolidate(self, messages, provider, model):
"""将投资对话压缩到长期记忆"""
# 1. 读取当前投资记忆
current_memory = self.read_long_term()
# 2. 构建提示词
prompt = f"""
## 当前投资记忆
{current_memory}
## 待处理的投资决策对话
{format_messages(messages)}
请提取关键信息：
- 新的持仓变化
- 调整的风险偏好
- 重要的投资洞察
- 止损/止盈点位更新
"""
# 3. 强制 LLM 调用 save_memory 工具
response = await provider.chat_with_retry(
messages=[{"role": "user", "content": prompt}],
tools=_SAVE_MEMORY_TOOL,
tool_choice={"type": "function", "function": {"name": "save_memory"}}
)
# 4. 提取结果
args = response.tool_calls[0].arguments
history_entry = args["history_entry"] # 交易摘要
memory_update = args["memory_update"] # 更新后的投资档案
# 5. 持久化
self.append_history(history_entry)
self.write_long_term(memory_update)Token 优化策略（投资者版）：
当会话超过上下文窗口一半时，自动触发记忆整合：
async def maybe_consolidate_by_tokens(self, session):
target = self.context_window_tokens // 2
estimated = estimate_session_prompt_tokens(session)
while estimated > target:
# 选择最早的未整合消息
boundary = pick_consolidation_boundary(session)
chunk = session.messages[session.last_consolidated:boundary]
# 压缩到记忆（保留关键投资决策）
await self.consolidate_messages(chunk)
# 更新偏移量
session.last_consolidated = boundary
estimated = estimate_session_prompt_tokens(session)这种设计确保了：
-
• ✅ 无限期跟踪：理论上可以永远记录你的投资历程 -
• ✅ 关键决策不丢失：重要买卖点、逻辑沉淀到长期记忆 -
• ✅ 高效检索：HISTORY.md 支持关键词搜索（如”NVDA 买入理由”） -
• ✅ 成本优化：减少重复信息的 token 消耗（持有个股只需记一次）
（3）心跳服务：自主投资引擎
nanobot/heartbeat/service.py 是实现 24 小时股市监控的核心：
class HeartbeatService:
"""周期性唤醒代理检查市场任务"""
def __init__(self, interval_s=30*60): # 默认 30 分钟
self.interval_s = interval_s
async def _run_loop(self):
while self._running:
await asyncio.sleep(self.interval_s)
await self._tick()
async def _tick(self):
# 1. 读取 HEARTBEAT.md（投资任务清单）
content = self._read_heartbeat_file()
# 2. 虚拟工具调用决策
action, tasks = await self._decide(content)
# 3. 如果有市场异动要处理
if action == "run":
response = await self.on_execute(tasks)
await self.on_notify(response) # 主动通知投资者HEARTBEAT.md 投资版示例：
# 投资监控任务
## 盘前监控 (美东时间 9:00)
- [ ] 检查期货市场：标普 500、纳斯达克、道指期货
- [ ] A50 期货异动预警
- [ ] 隔夜中概股表现
- [ ] 财经日历：今日财报、经济数据发布
## 盘中监控 (每 30 分钟)
- [ ] 持仓股票价格波动超 3% 立即分析
- [ ] 成交量异常放大（>5 日均量 2 倍）
- [ ] 板块轮动监测：资金流向
- [ ] 期权 Put/Call Ratio 异常
## 盘后复盘 (美东时间 17:00)
- [ ] 生成当日交易总结
- [ ] 更新技术指标（MACD、KDJ、RSI）
- [ ] 龙虎榜数据分析
- [ ] 准备明日交易计划
## 特殊事件驱动
- [ ] 美联储议息会议实时解读
- [ ] 财报电话会议要点整理
- [ ] 突发新闻情感分析（战争、政策、监管）决策工具定义：
_HEARTBEAT_TOOL = [{
"type": "function",
"function": {
"name": "heartbeat",
"parameters": {
"action": {"enum": ["skip", "run"]},
"tasks": "任务描述（run 时必须）"
}
}
}]这种设计的精妙之处：
-
• 🎯 避免无效打扰：大部分心跳返回 “skip”，只在关键时刻唤醒 -
• 🎯 紧急事件优先：重大异动（如闪崩、暴涨）立即推送 -
• 🎯 可解释性强：每个操作都有明确的逻辑依据 -
• 🎯 灵活配置：根据投资者作息调整监控频率（如夜猫子模式）
（4）定时任务：投资策略自动执行
nanobot/cron/service.py 提供强大的投资任务调度能力：
class CronService:
""" cron 任务调度服务（投资者版）"""
def add_job(
self,
name: str,
schedule: CronSchedule,
message: str,
deliver: bool = False,
):
"""添加定时投资任务"""
# 支持三种调度模式
schedule = CronSchedule(
kind="cron", # cron 表达式
expr="0 9 * * *", # 每天早上 9 点（股市开盘）
tz="America/New_York"
)
# 或
schedule = CronSchedule(
kind="every", # 固定间隔
every_ms=1800000 # 每 30 分钟（盘中监控）
)
# 或
schedule = CronSchedule(
kind="at", # 一次性时间点
at_ms=timestamp # 财报发布时刻
)真实投资场景配置：
{
"jobs": [
{
"name": "早盘前瞻",
"schedule": {
"kind": "cron",
"expr": "30 9 * * *",
"tz": "Asia/Shanghai"
},
"message": "分析隔夜美股、A50 期货、人民币汇率，给出今日策略建议"
},
{
"name": "持仓监控",
"schedule": {
"kind": "every",
"every_ms": 1800000 // 30 分钟
},
"message": "检查持仓股票价格和成交量，发现异常波动立即分析原因并推送"
},
{
"name": "周报生成",
"schedule": {
"kind": "cron",
"expr": "0 20 * * 5", // 周五晚 8 点
"tz": "Asia/Shanghai"
},
"message": "整理本周交易记录、收益率、胜率，生成 Markdown 报告并对比基准指数"
},
{
"name": "财报季提醒",
"schedule": {
"kind": "cron",
"expr": "0 8 * * *",
"tz": "America/New_York"
},
"message": "检查持仓公司今日是否发布财报，如有则整理市场预期和历史表现"
}
]
}执行流程：
async def _execute_job(self, job):
logger.info(f"执行投资任务：{job.name}")
try:
# 调用 agent loop 处理任务
response = await self.on_job(job)
job.state.last_status = "ok"
# 如果需要主动推送给投资者
if job.payload.deliver:
await send_to_channel(
channel=job.payload.channel,
to=job.payload.to,
content=response
)
except Exception as e:
job.state.last_status = "error"
job.state.last_error = str(e)
# 计算下次执行时间
job.state.next_run_at_ms = _compute_next_run(job.schedule)（5）工具注册表：投资分析武器库
nanobot/agent/tools/registry.py 实现动态工具管理，投资者可扩展自己的分析工具箱：
class ToolRegistry:
"""投资工具注册表"""
def __init__(self):
self._tools = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
async def execute(self, name, params):
tool = self._tools.get(name)
# 参数验证
errors = tool.validate_params(params)
if errors:
return f"Error: {errors}"
# 执行工具
result = await tool.execute(**params)
return result内置投资工具清单：
|
|
|
|
|---|---|---|
get_stock_price |
|
|
get_financials |
|
|
calculate_indicators |
|
|
fetch_news |
|
|
sentiment_analysis |
|
|
screen_stocks |
|
|
backtest_strategy |
|
|
portfolio_rebalance |
|
|
options_flow |
|
|
insider_trading |
|
|
自定义投资工具示例：
# custom_tools.py
from nanobot.agent.tools.base import Tool
import yfinance as yf
class StockPriceTool(Tool):
name = "get_stock_price"
description = "获取股票实时价格和历史走势"
parameters = {
"symbol": {"type": "string", "description": "股票代码，如 AAPL, TSLA, 00700.HK"},
"period": {"type": "string", "description": "时间范围：1d, 5d, 1mo, 3mo, 1y", "default": "1d"}
}
async def execute(self, symbol: str, period: str = "1d") -> str:
ticker = yf.Ticker(symbol)
hist = ticker.history(period=period)
current_price = hist['Close'].iloc[-1]
change_pct = (current_price - hist['Open'].iloc[0]) / hist['Open'].iloc[0] * 100
return f"{symbol} 当前价格：${current_price:.2f} ({change_pct:+.2f}%)\n\n最高：${hist['High'].max():.2f}\n最低：${hist['Low'].min():.2f}\n成交量：{hist['Volume'].sum():,.0f}"
# 注册工具
registry.register(StockPriceTool())MCP 集成（连接专业数据源）：
# 配置 MCP 服务器
{
"tools": {
"mcpServers": {
"bloomberg-terminal": {
"url": "https://mcp.bloomberg.com/v1",
"headers": {"Authorization": "Bearer YOUR_API_KEY"}
},
"wind-financial": {
"command": "npx",
"args": ["@windfinancial/mcp-server"],
"env": {"WIND_API_KEY": "xxx"}
}
}
}
}MCP 工具会自动包装为 nanobot 工具：
-
• mcp_bloomberg_get_quote -
• mcp_wind_financial_get_fundamentals
这使得 nanobot 可以：
-
• 🔌 连接专业终端：Bloomberg、Wind、Refinitiv 数据直达 -
• 🔌 券商 API 集成：Interactive Brokers、富途、老虎证券自动下单 -
• 🔌 安全沙箱：工具执行在独立进程中，交易指令双重确认
（6）通道抽象：统一消息接口
nanobot/channels/base.py 定义了标准化的通道接口：
class BaseChannel(ABC):
"""通道基类"""
@abstractmethod
async def start(self):
"""启动通道，监听消息"""
pass
@abstractmethod
async def send(self, msg: OutboundMessage):
"""发送消息"""
pass
async def _handle_message(self, sender_id, chat_id, content, media=None):
"""处理收到的消息并发布到总线"""
# 权限检查
if not self.is_allowed(sender_id):
logger.warning("访问拒绝")
return
# 发布到消息总线
msg = InboundMessage(
channel=self.name,
sender_id=sender_id,
chat_id=chat_id,
content=content,
media=media
)
await self.bus.publish_inbound(msg)具体实现示例（Telegram）：
class TelegramChannel(BaseChannel):
name = "telegram"
display_name = "Telegram"
async def start(self):
from telegram import Update
from telegram.ext import Application, MessageHandler, filters
app = Application.builder().token(self.config.token).build()
async def handle_message(update, context):
text = update.message.text
audio = update.message.voice
# 语音转文字
if audio:
file = await context.bot.get_file(audio.file_id)
audio_path = await file.download_to_drive()
text = await self.transcribe_audio(audio_path)
# 转发到 agent loop
await self._handle_message(
sender_id=str(update.message.from_user.id),
chat_id=str(update.message.chat_id),
content=text
)
app.add_handler(
MessageHandler(filters.ALL, handle_message)
)
await app.start_polling()目前已实现的通道：
|
|
|
|
|---|---|---|
| Telegram |
|
|
| Discord |
|
|
|
|
|
|
| 飞书 |
|
|
| 钉钉 |
|
|
| Slack |
|
|
|
|
|
|
|
|
|
|
| 企业微信 |
|
|
四、相比 OpenClaw 的优势
4.1 代码复杂度对比
OpenClaw:
├── 50,000+ 行代码
├── 多层抽象（Framework → Core → Plugins）
├── 依赖注入容器
├── 事件总线系统
├── 插件热加载机制
└── 学习曲线：⭐⭐⭐⭐⭐
Nanobot:
├── ~500 行核心代码
├── 扁平化设计（Agent Loop + Tools + Memory）
├── 配置即代码
├── 单一职责原则
└── 学习曲线：⭐⭐4.2 性能指标实测
|
|
|
|
|
|---|---|---|---|
| 冷启动时间 |
|
|
10-20x |
| 内存占用 |
|
|
4-5x |
| 首次响应延迟 |
|
|
3x |
| 并发会话数 |
|
|
|
4.3 开发效率对比
添加一个新工具：
OpenClaw：
# 1. 定义工具接口
@plugin_interface
class MyTool(Plugin):
def __init__(self, container: DIContainer):
super().__init__(container)
self.event_bus = container.get(EventBus)
self.config = container.get(Config)
@override
async def initialize(self):
# 初始化逻辑...
pass
@plugin_method
async def execute(self, params: ToolParams) -> ToolResult:
# 业务逻辑...
passNanobot：
# 1. 继承 Tool 基类
class MyTool(Tool):
name = "my_tool"
description = "Do something useful"
parameters = {"query": {"type": "string"}}
async def execute(self, query: str) -> str:
# 直接写业务逻辑
result = await do_something(query)
return result代码量对比：OpenClaw 需要 50+ 行模板代码，Nanobot 仅需 10 行。
4.4 部署复杂度
OpenClaw 部署流程：
# 1. 准备 Docker Compose
docker-compose up -d redis postgres elasticsearch
# 2. 配置环境变量
export OPENCLAW_REDIS_URL=redis://localhost:6379
export OPENCLAW_DB_URL=postgresql://user:pass@localhost/db
# 3. 运行迁移
openclaw migrate
# 4. 启动服务
openclaw start --workers 4Nanobot 部署流程：
# 1. 安装包
pip install nanobot-ai
# 2. 配置 API Key
nano ~/.nanobot/config.json
# 3. 启动
nanobot gateway一句话总结：OpenClaw 是企业级航母，Nanobot 是个人快艇——前者功能全面但笨重，后者灵活敏捷易上手。
五、对普通人的价值
5.1 开发者：提效神器
场景 1：代码审查助手
# 配置每天上午 10 点的定时任务
nanobot cron add "代码审查" \
--cron "0 10 * * *" \
--message "检查昨天提交的代码，找出潜在 bug 和改进点，生成报告"第二天早上 10 点，你会收到：
📊 代码审查报告 - 2026-03-18
## 发现的问题
1. ⚠️ 空指针风险：user_service.py:45
- 未检查 user 是否为 None
2. 🐛 性能瓶颈：data_loader.py:123
- 循环内查询数据库，建议批量加载
## 改进建议
- 添加类型注解（PEP 484）
- 使用 logging 替代 print
- 重构嵌套过深的函数场景 2：自动化运维
{
"heartbeat": {
"interval": "5m",
"tasks": [
"监控服务器 CPU 使用率，超过 80% 发出警告",
"检查磁盘空间，剩余少于 10GB 时清理日志",
"查看错误日志，统计最近的异常"
]
}
}场景 3：学习伙伴
# 通过 Telegram 向 bot 发送
"帮我理解 Rust 的所有权机制，用 Python 程序员能懂的方式解释"
# Bot 回复：
"""
Rust 所有权就像 Python 的引用计数 + 作用域规则：
1. 每个值有一个"所有者"变量（类似 Python 变量绑定）
2. 同一时间只能有一个所有者（Rust 独有，Python 可多引用）
3. 所有者离开作用域，值被丢弃（类似 Python GC，但确定性更强）
示例对比：
Python:
>>> a = [1, 2, 3]
>>> b = a # a 和 b 共享同一个列表
>>> b.append(4)
>>> print(a) # [1, 2, 3, 4] - 都被修改了
Rust:
let a = vec![1, 2, 3];
let b = a; // 所有权转移给 b，a 失效
// println!("{:?}", a); // 编译错误！
b.push(4); // 只有 b 能修改5.2 非技术人员：生活助理
场景 1：信息聚合
{
"cron_jobs": [
{
"name": "每日早报",
"schedule": {"kind": "cron", "expr": "0 8 * * *"},
"message": "总结昨天的科技新闻、股市行情、天气预报"
}
]
}每天早上 8 点收到：
☀️ 早安！今天是 2026 年 3 月 18 日 星期三
## 📰 科技新闻
- OpenAI 发布 GPT-5，推理能力提升 40%
- 特斯拉新款 Model Y 续航突破 700km
## 📈 股市概览
- 上证指数：3050 点 (+1.2%)
- 纳斯达克：15800 点 (+0.8%)
## 🌤️ 今日天气
北京：晴，15-25°C，适宜出行场景 2：健康管理
# 设置每小时提醒
nanobot cron add "喝水提醒" \
--every 1h \
--message "提醒我站起来走动、喝水、远眺放松眼睛"场景 3：财务监控
# 心跳任务配置
HEARTBEAT.md:
- 每 30 分钟检查基金净值
- 如果单日跌幅超过 3%，发送安慰和建议
- 每周五生成投资周报5.3 学生群体：学习加速器
场景 1：论文阅读助手
# 上传 PDF 到工作目录
nanobot agent "总结这篇论文的核心贡献，用通俗语言解释方法论"场景 2：编程辅导
学生："这段 Python 代码为什么报错？"
[粘贴代码]
Nanobot:
"""
问题在第 15 行：你试图访问字典中不存在的键。
错误原因：
scores = {}
print(scores['math']) # KeyError: 'math'
解决方案：
1. 使用 .get() 方法：scores.get('math', 0)
2. 先检查键是否存在：if 'math' in scores:
3. 使用 defaultdict: from collections import defaultdict
推荐方案 1，代码最简洁。
"""场景 3：语言学习
"我想练习雅思口语，请扮演考官，随机提问并给我的回答打分"
Nanobot:
"""
好的，我是你的雅思口语考官。今天我们讨论话题：Technology
Question 1: How often do you use your smartphone?
（请用英语回答，我会从流利度、词汇、语法、发音四个维度评分）
"""5.4 银发族：数字伴侣
虽然 nanobot 目前主要面向技术人员，但其设计理念让所有人都能受益：
-
• 语音交互：通过 Telegram 语音消息，自动转录为文字 -
• 简单配置：子女帮忙设置一次，父母长期使用 -
• 主动关怀：定时提醒吃药、运动、联系家人 -
• 信息查询：问天气、查新闻、找菜谱
六、实战演练：从零搭建 24 小时股市监控系统
6.1 快速开始（10 分钟上手）
步骤 1：安装
# 方式 1：PyPI（稳定版）
pip install nanobot-ai yfinance pandas ta-lib
# 方式 2：源码（最新版）
git clone https://github.com/HKUDS/nanobot.git
cd nanobot
pip install -e .步骤 2：初始化
nanobot onboard这会自动创建配置文件 ~/.nanobot/config.json。
步骤 3：配置投资助理
编辑 ~/.nanobot/config.json：
{
"providers": {
"openrouter": {
"apiKey": "sk-or-v1-xxxxxxxxxxxxx"
}
},
"agents": {
"defaults": {
"model": "anthropic/claude-opus-4-5",
"provider": "openrouter"
}
},
"tools": {
"custom_tools": [
"yfinance_price",
"financial_news",
"technical_indicators"
]
}
}步骤 4：创建投资心跳文件
创建工作目录并创建 HEARTBEAT.md：
mkdir ~/investment-bot
cd ~/investment-bot
cat > HEARTBEAT.md << 'EOF'
# 投资监控任务清单
## 持仓股票
- AAPL (苹果) - 成本 $175
- TSLA (特斯拉) - 成本 $220
- NVDA (英伟达) - 成本 $480
- 00700.HK (腾讯) - 成本 HKD 350
## 监控规则
### 价格预警
- 单日涨跌幅超过 5% → 立即分析原因并推送
- 突破 20 日新高 → 评估是否加仓
- 跌破 60 日均线 → 提醒止损风险
### 财报季
- 持仓公司发布财报 → 1 小时内整理要点
- 电话会议记录 → 提取管理层指引
- 卖方分析师点评 → 汇总多空观点
### 宏观事件
- 美联储议息会议 → 实时解读点阵图
- CPI/PCE 数据发布 → 分析通胀趋势
- 地缘政治突发 → 评估对市场影响
## 报告生成
- 每日收盘后：交易日报（盈亏、操作、明日计划）
- 每周五晚：周度复盘（胜率、最大回撤、改进点）
- 每月末：月度总结（收益率对比基准、仓位调整）
EOF步骤 5：CLI 测试
nanobot agent
> 帮我分析 TSLA 今天的走势，有什么重要消息吗？
正在获取 TSLA 实时价格... ✅ $238.45 (+3.2%)
正在查询成交量... ✅ 1.25 亿股 (5 日均量 1.8 倍)
正在搜索相关新闻... ✅ 找到 15 条
📊 TSLA 今日分析
## 价格表现
当前价：$238.45 (+3.2%)
日内区间：$232.10 - $240.80
成交量：1.25 亿股 (放量 80%)
## 驱动因素
1. 🚗 Q3 交付量超预期：43.5 万辆 vs 预期 42 万辆
2. 🔋 上海工厂产能提升：周产 6,500 辆创新高
3. 🤖 FSD V12 好评率 95%，订阅用户增长 30%
## 技术面
- MACD 金叉第 3 天，红柱放大
- RSI: 62 (未超买)
- 阻力位：$245 (前高)
- 支撑位：$230 (5 日线)
## 操作建议
✅ 持有：上升趋势完好，目标看至$250
⚠️ 如突破$245 可考虑加仓
❌ 止损位下移至$225步骤 6：连接微信/Telegram（可选）
-
1. 在 Telegram 搜索 @BotFather -
2. 发送 /newbot创建机器人 -
3. 复制 Token
更新配置：
{
"channels": {
"telegram": {
"enabled":true,
"token": "YOUR_BOT_TOKEN",
"allowFrom": ["YOUR_USER_ID"]
}
}
}启动网关：
nanobot gateway现在可以在 Telegram 上随时接收投资预警了！
6.2 进阶配置：24 小时监控系统
目标：搭建一个实时分析加密货币市场的 AI 助理。
步骤 1：创建工作目录
mkdir ~/crypto-bot
cd ~/crypto-bot步骤 2：配置心跳任务
创建 HEARTBEAT.md：
# 加密货币监控任务
## 监控目标
- BTC/USDT 价格
- ETH/USDT 价格
- 市场恐慌贪婪指数
## 预警阈值
- BTC 单日涨跌幅超过 5%
- ETH 单日涨跌幅超过 7%
- 恐慌指数低于 20 或高于 80
## 执行动作
1. 触发预警时，立即发送 Telegram 消息
2. 分析原因（搜索新闻）
3. 给出操作建议（持有/买入/卖出）步骤 3：配置定时任务
编辑 ~/.nanobot/config.json：
{
"cron": {
"enabled":true,
"jobs": [
{
"name": "价格检查",
"schedule": {
"kind": "every",
"every_ms": 1800000 // 30 分钟
},
"message": "检查 BTC 和 ETH 的价格，计算 24 小时涨跌幅。如果发现异常波动（>5%），立即搜索相关新闻并分析原因。"
},
{
"name": "日报生成",
"schedule": {
"kind": "cron",
"expr": "0 20 * * *", // 晚上 8 点
"tz": "Asia/Shanghai"
},
"message": "总结今天的市场行情，绘制价格走势图（保存为图片），列出影响市场的关键事件。"
}
]
}
}步骤 4：添加工具（可选）
如果想直接获取实时价格，可以添加自定义工具：
# crypto_tool.py
from nanobot.agent.tools.base import Tool
import httpx
class CryptoPriceTool(Tool):
name = "crypto_price"
description = "获取加密货币实时价格"
parameters = {
"symbol": {"type": "string", "description": "币种符号，如 BTC, ETH"}
}
async def execute(self, symbol: str) -> str:
url = f"https://api.binance.com/api/v3/ticker/price"
params = {"symbol": f"{symbol}USDT"}
async with httpx.AsyncClient() as client:
resp = await client.get(url, params=params)
data = resp.json()
price = float(data["price"])
return f"{symbol} 当前价格：${price:,.2f} USDT"
# 注册工具
# 在启动脚本中添加
registry.register(CryptoPriceTool())步骤 5：启动服务
# 终端 1：启动网关（监听 Telegram 消息）
nanobot gateway
# 终端 2：查看状态
nanobot status预期效果：
-
• 每 30 分钟自动检查价格 -
• 发现暴涨暴跌立即推送警报 -
• 每晚 8 点发送日报总结 -
• 随时可以通过 Telegram 询问当前价格
6.3 高级玩法：多代理协作
场景：组建一个 AI 团队，分别负责研究、写作、审核。
# 主代理：项目经理
"我们需要写一篇关于量子计算的技术博客。我来协调分工。"
# spawned 子代理 1：研究员
spawn("researcher", """
调研量子计算的最新进展：
1. 谷歌、IBM 的最新量子处理器
2. 量子优势的实验验证
3. 潜在的应用场景
""")
# spawned 子代理 2：作家
spawn("writer", """
根据研究员提供的资料，撰写博客文章：
- 引言：为什么量子计算重要
- 主体：技术突破详解
- 结尾：对未来影响
要求：通俗易懂，避免过多数学公式
""")
# spawned 子代理 3：审核员
spawn("reviewer", """
审核文章：
1. 事实准确性检查
2. 逻辑连贯性
3. 可读性改进建议
""")
# 主代理汇总
"感谢各位的工作！最终文章如下：[发布]"这种多代理模式通过 spawn 工具实现：
class SpawnTool(Tool):
name = "spawn"
description = "创建子代理处理并行任务"
parameters = {
"role": {"type": "string", "description": "角色名称"},
"task": {"type": "string", "description": "任务描述"}
}
async def execute(self, role: str, task: str) -> str:
# 创建独立的 AgentLoop 实例
subagent = await self.manager.create_subagent(
role=role,
task=task,
parent_session=self.session_key
)
# 等待子代理完成
result = await subagent.run(task)
return result七、技术挑战与解决方案
7.1 长上下文管理
问题：LLM 的上下文窗口有限（通常 128K tokens），长时间对话会超出限制。
Nanobot 方案：双层记忆 + 动态压缩
# 估算当前 token 数
estimated = estimate_prompt_tokens_chain(
provider, model, messages, tools
)
# 如果超过阈值（50% 窗口）
if estimated > context_window_tokens // 2:
# 选择最早的消息进行压缩
boundary = pick_consolidation_boundary(session)
chunk = session.messages[:boundary]
# 调用 LLM 提炼关键信息
await consolidate_messages(chunk)
# 从会话中移除已压缩的消息
session.last_consolidated = boundary效果：理论上支持无限长度的对话，同时保留关键信息。
7.2 并发控制
问题：多个用户同时发送消息，如何处理？
Nanobot 方案：会话锁 + 任务队列
class AgentLoop:
def __init__(self):
self._processing_lock = asyncio.Lock()
self._active_tasks = {} # session_key -> tasks
async def _dispatch(self, msg):
# 全局锁保证同一会话串行处理
async with self._processing_lock:
try:
response = await self._process_message(msg)
except Exception:
logger.exception("处理失败")
async def _handle_stop(self, msg):
# 取消该会话的所有任务
tasks = self._active_tasks.pop(msg.session_key, [])
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)7.3 错误恢复
问题：工具调用失败、LLM 返回错误、网络超时怎么办？
Nanobot 方案：多层次容错
# 1. 工具执行错误：添加提示引导 LLM 重试
async def execute(self, name, params):
try:
result = await tool.execute(**params)
if result.startswith("Error"):
return result + "\n\n[Analyze the error above and try a different approach.]"
except Exception as e:
return f"Error: {e}\n\n[Analyze the error and try a different approach.]"
# 2. LLM 错误：降级处理
if response.finish_reason == "error":
logger.error("LLM 返回错误")
final_content = "抱歉，遇到错误，请稍后再试"
break # 跳出循环，避免死循环
# 3. 记忆整合失败：降级为原始归档
def _fail_or_raw_archive(self, messages):
self._consecutive_failures += 1
if self._consecutive_failures >= 3:
# 连续失败 3 次后，直接保存原始消息
self._raw_archive(messages)
self._consecutive_failures = 07.4 安全性
问题：如何防止恶意用户执行危险命令？
Nanobot 方案：多重防护
# 1. 访问控制
class BaseChannel:
def is_allowed(self, sender_id):
allow_list = self.config.allow_from
if "*" in allow_list:
return True # 允许所有人
return sender_id in allow_list # 白名单
# 2. 工作目录限制
class ExecTool(Tool):
def __init__(self, restrict_to_workspace=False):
self.restrict_to_workspace = restrict_to_workspace
self.working_dir = Path.cwd()
async def execute(self, command: str):
if self.restrict_to_workspace:
# 禁止 cd .. 跳出工作目录
if ".." in command:
return "Error: 不允许离开工作目录"
# 3. 超时保护
try:
result = await asyncio.wait_for(
run_command(command),
timeout=self.timeout
)
except asyncio.TimeoutError:
return "Error: 命令执行超时"八、生态与扩展
8.1 技能市场
Nanobot 支持从 ClawHub 安装第三方技能：
# 搜索技能
nanobot clawhub search "weather"
# 安装技能
nanobot clawhub install weather-skill
# 使用技能
"查询北京明天的天气"已发布的技能包括：
-
• 🌤️ 天气预报 -
• 📊 股票市场 -
• 📝 文档总结 -
• 🧠 记忆增强 -
• ⏰ 智能提醒 -
• 🔍 GitHub 趋势
8.2 自定义 Provider
除了官方支持的模型，还可以自定义 Provider：
# custom_provider.py
from nanobot.providers.base import LLMProvider
class MyCustomProvider(LLMProvider):
async def chat_with_retry(self, messages, tools, model):
# 调用你自己的 LLM API
response = await my_llm_api.chat(
messages=messages,
tools=tools,
model=model
)
return response配置：
{
"providers": {
"custom": {
"apiKey": "your-key",
"apiBase": "https://api.your-llm.com/v1"
}
}
}8.3 社区资源
-
• GitHub: https://github.com/HKUDS/nanobot -
• Discord: https://discord.gg/MnCvHqpUGB -
• 文档: https://nanobot.dev -
• 技能库: https://clawhub.ai
九、未来展望
9.1 路线图
根据 README 中的更新日志，nanobot 正在快速发展：
短期（2026 Q2）：
-
•更多通道支持（Line、KakaoTalk） -
•视觉模型集成（图像理解） -
•语音通话功能（实时语音交互） -
•多语言支持（中文、日语、韩语优化）
中期（2026 Q3-Q4）：
-
•分布式部署（多实例协同） -
•边缘计算支持（离线模式） -
•插件市场（一键安装技能） -
•可视化配置界面
长期（2027+）：
-
•自主目标设定（AGI 雏形） -
•情感计算（共情能力） -
•群体智能（多 bot 协作网络） -
•脑机接口探索
9.2 愿景
Nanobot 的终极目标不是做一个”更好的聊天机器人”，而是：
让每个人都拥有一个专属的 AI 助理，像《钢铁侠》中的 JARVIS 一样，24 小时忠诚服务，不断学习和成长。
这个愿景包含三个层次：
-
1. 可用性：普通人也能轻松部署和使用 -
2. 个性化：了解你的习惯、偏好、知识背景 -
3. 自主性：不仅能回答问题，还能主动发现问题、解决问题
十、结语
在这个 AI 技术日新月异的时代，nanobot 代表了一种新的可能性：
-
• 不是大公司的专利：无需昂贵的基础设施，树莓派就能运行 -
• 不是研究者的玩具：真实可用的 24 小时助理，而非演示 demo -
• 不是极客的专属：简单的配置，友好的交互，人人都能用
正如 nanobot 的名字所暗示的：
Nano – 微小但强大
Bot – 你的忠实伙伴
它可能不像 GPT-5 那样无所不知，但它专属于你，了解你的工作流，记住你的偏好，在你需要时及时出现，在你忙碌时默默守护。
这就是 nanobot 的意义：AI 不应该只是云端的服务，而应该是每个人触手可及的日常工具。
附录：快速参考卡片
A. 常用命令
# 安装
pip install nanobot-ai
# 初始化
nanobot onboard
# 启动 CLI
nanobot agent
# 启动网关（多通道）
nanobot gateway
# 查看状态
nanobot status
# 添加定时任务
nanobot cron add "任务名" --cron "0 9 * * *" --message "任务描述"
# 安装技能
nanobot clawhub install skill-name
# 查看日志
tail -f ~/.nanobot/logs/nanobot.logB. 配置文件模板
{
"providers": {
"openrouter": {"apiKey": "sk-or-xxx"},
"groq": {"apiKey": "gsk_xxx"} // 用于语音转录
},
"agents": {
"defaults": {
"model": "anthropic/claude-opus-4-5",
"provider": "openrouter"
}
},
"channels": {
"telegram": {
"enabled":true,
"token": "BOT_TOKEN",
"allowFrom": ["USER_ID"]
}
},
"cron": {
"enabled":true,
"jobs": [...]
},
"heartbeat": {
"enabled":true,
"interval_s": 1800
}
}C. 故障排查
|
|
|
|
|---|---|---|
|
|
|
config.json |
|
|
|
|
|
|
|
~/.nanobot/sessions |
|
|
|
|
参考文献：
-
1. Nanobot 官方文档：https://github.com/HKUDS/nanobot -
2. OpenClaw 项目：https://github.com/openclaw/openclaw -
3. LiteLLM 文档：https://docs.litellm.ai -
4. MCP 协议规范：https://modelcontextprotocol.io -
5. Croniter 库：https://github.com/kiorky/croniter
关于作者：
本文由 AI 辅助完成，审校者为一位 nanobot 的深度用户。如果你也在使用 nanobot，欢迎加入社区分享你的经验！
如果你觉得这篇文章有帮助，请点赞、收藏、转发，让更多人受益于 nanobot！
