快速开始

# 推荐用 venv 隔离环境
python -m venv muyuan-env
source muyuan-env/bin/activate  # Windows: muyuan-env\Scripts\activate

# 安装
pip install muyuan-ai

muyuan --version
# 输出: muyuan-ai 0.3.0

git clone https://github.com/muyuan-ai/muyuan-ai.git
cd muyuan-ai
pip install -e ".[dev]"  # 安装开发依赖
pytest                    # 跑一遍测试确认环境没问题

from muyuan import Collector, Analyzer, Advisor

# 1. 初始化收集器，指定数据源和过滤条件
collector = Collector(
    sources=["sina_news", "eastmoney"],
    filters={
        "sector": "technology",
        "min_relevance": 0.6
    }
)

# 2. 抓取数据
data = collector.fetch(period="24h")
print(f"抓到 {len(data.articles)} 篇文章")

# 3. 分析
insights = Analyzer.run(data, model="sentiment_v2")

# 4. 生成建议
advice = Advisor.generate(
    insights,
    risk_level="moderate",
    horizon="medium_term"
)

# 5. 输出
print(advice.summary)
print("---")
for rec in advice.recommendations:
    print(f"[{rec.confidence:.0%}] {rec.action} {rec.target} — {rec.reason}")

抓到 47 篇文章
科技板块短期承压，但 AI 基建方向中长期确定性高。
---
[82%] 关注 算力基础设施 — 三大运营商资本开支计划超预期
[71%] 观望 消费电子 — 需求复苏信号不明确，等待 Q2 财报
[65%] 减持 在线教育 — 政策风险仍在释放中

# config.yaml — 目远AI 配置文件

app:
  name: "my-muyuan-instance"
  log_level: "info"          # debug / info / warning / error
  cache_dir: "~/.muyuan/cache"

collector:
  default_sources:
    - "sina_news"
    - "eastmoney"
    - "xueqiu"
  fetch_interval: 3600       # 抓取间隔，单位秒
  max_articles: 500          # 单次最大抓取量
  proxy: null                  # HTTP 代理，格式 "http://host:port"

analyzer:
  model: "sentiment_v2"       # 分析模型版本
  language: "zh"              # zh / en / auto
  min_confidence: 0.5        # 低于此阈值的分析结果会被丢弃

advisor:
  risk_level: "moderate"      # conservative / moderate / aggressive
  horizon: "medium_term"     # short_term / medium_term / long_term
  max_recommendations: 10

database:
  url: "postgresql://localhost:5432/muyuan"
  pool_size: 5

redis:
  url: "redis://localhost:6379/0"
  ttl: 7200                   # 缓存过期时间，单位秒

from muyuan.collectors.base import BaseCollector, Article
from datetime import datetime
import httpx

class MyRSSCollector(BaseCollector):
    """从 RSS 源抓取文章"""

    name = "my_rss"
    requires_api_key = False

    def __init__(self, feed_url: str):
        self.feed_url = feed_url

    async def fetch(self, since: datetime) -> list[Article]:
        async with httpx.AsyncClient() as client:
            resp = await client.get(self.feed_url)
            entries = self._parse_feed(resp.text)

        return [
            Article(
                title=e["title"],
                content=e["summary"],
                source=self.name,
                published_at=e["published"],
                url=e["link"],
            )
            for e in entries
            if e["published"] >= since
        ]

collector:
  custom_sources:
    - module: "my_collectors.rss"
      class: "MyRSSCollector"
      args:
        feed_url: "https://example.com/finance/rss.xml"

secrets:
  tushare_token: "你的 token"