AI 的正確位置:訊號層,不是執行層

建立情緒評分 Script

先安裝需要的套件:

pip install anthropic requests pandas

建立 sentiment_scorer.py,這個 script 的工作是抓新聞標題、呼叫 Claude API 評分、把結果附加到 CSV:

# sentiment_scorer.py
import anthropic
import json
import pandas as pd
import requests
from datetime import datetime, timezone, timedelta
from pathlib import Path

# 使用 Claude Haiku 4.5 — 情緒分類任務用不到頂尖模型,省成本
ANTHROPIC_API_KEY = "your_api_key_here"  # 實際使用時從環境變數讀取
client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

SENTIMENT_CSV = Path("user_data/sentiment_btc.csv")

def score_news(headline: str) -> dict:
    """
    用 Claude Haiku 評估單篇新聞對 BTC 的情緒分數。
    回傳 dict:{"score": float, "reason": str}
    """
    prompt = f"""你是加密貨幣市場情緒分析師,只需要評估下面這則新聞對 BTC 市場的短期影響。

新聞標題:{headline}

請以 JSON 格式回覆,只輸出 JSON,不加其他文字:
{{"score": 0.0, "reason": "一句話說明"}}

score 評分規則:
 1.0 = 極度看漲(ETF 核准、主要機構大量採購等)
 0.5 = 溫和看漲
 0.0 = 中性或無關
-0.5 = 溫和看跌
-1.0 = 極度看跌(重大交易所爆雷、嚴厲監管打壓等)"""

    try:
        msg = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=120,
            messages=[{"role": "user", "content": prompt}]
        )
        return json.loads(msg.content[0].text)
    except (json.JSONDecodeError, Exception):
        # API 呼叫失敗或解析失敗 → 回傳中性分數,不讓策略因此停擺
        return {"score": 0.0, "reason": "parse_error"}


def fetch_crypto_headlines() -> list[str]:
    """
    從 CryptoCompare 免費 API 抓最新 BTC 相關新聞標題。
    免費方案每月 10 萬次呼叫,每小時跑一次完全足夠。
    """
    url = "https://min-api.cryptocompare.com/data/v2/news/?categories=BTC&excludeCategories=Sponsored&lang=EN"
    try:
        resp = requests.get(url, timeout=10)
        data = resp.json()
        # 只取最近 6 小時的新聞
        cutoff = datetime.now(timezone.utc) - timedelta(hours=6)
        return [
            item["title"]
            for item in data.get("Data", [])
            if datetime.fromtimestamp(item["published_on"], tz=timezone.utc) > cutoff
        ][:20]  # 最多 20 篇,控制成本
    except Exception:
        return []


def run_scoring():
    """主流程:抓新聞 → 評分 → 算平均 → 寫入 CSV"""
    headlines = fetch_crypto_headlines()
    if not headlines:
        print("沒抓到新聞,跳過這次評分")
        return

    scores = []
    for headline in headlines:
        result = score_news(headline)
        scores.append(result["score"])
        print(f"[{result['score']:+.1f}] {headline[:60]}")

    avg_score = sum(scores) / len(scores)
    now_hour = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)

    row = pd.DataFrame([{
        "date": now_hour.strftime("%Y-%m-%d %H:00:00"),
        "sentiment_score": round(avg_score, 4),
        "news_count": len(scores)
    }])

    if SENTIMENT_CSV.exists():
        existing = pd.read_csv(SENTIMENT_CSV)
        # 去除同一小時的重複紀錄
        existing = existing[existing["date"] != row["date"].iloc[0]]
        df = pd.concat([existing, row], ignore_index=True)
    else:
        SENTIMENT_CSV.parent.mkdir(parents=True, exist_ok=True)
        df = row

    df.to_csv(SENTIMENT_CSV, index=False)
    print(f"\n平均情緒分數:{avg_score:+.4f}  (共 {len(scores)} 篇新聞)")


if __name__ == "__main__":
    run_scoring()

跑一次測試:

python sentiment_scorer.py

正常應該會看到每篇新聞的分數列出,以及最後的平均值和 CSV 更新訊息。把這個 script 設定成每小時自動執行(macOS 用 crontab,Linux 用 systemd timer 或 cron):

# crontab -e 加入這一行:每小時整點執行
0 * * * * /path/to/freqtrade-env/bin/python /path/to/my-quant/sentiment_scorer.py

在 freqtrade 策略中讀取情緒分數

開啟第 4 課建立的策略檔案(或新建一個繼承 IStrategy 的策略)。我們在 __init__ 預先載入情緒 CSV,在 populate_indicators 合併進 DataFrame:

# my_quant/user_data/strategies/SentimentStrategy.py
import pandas as pd
from pathlib import Path
from freqtrade.strategy import IStrategy, IntParameter
import talib.abstract as ta


class SentimentStrategy(IStrategy):
    """
    基礎均線 + 動量策略,加上 LLM 新聞情緒過濾器。
    情緒分數低於閾值時不進場,避免在負面情緒環境做多。
    """

    # 策略基本設定
    timeframe = "1h"
    stoploss = -0.05
    trailing_stop = False
    can_short = False

    # 情緒過濾閾值:可用 hyperopt 優化,先手動設定
    sentiment_threshold = -0.2  # 情緒分數低於 -0.2 時禁止做多

    def __init__(self, config: dict):
        super().__init__(config)
        # 預先載入情緒 CSV,讓每次 populate_indicators 不用重新讀檔
        sentiment_path = Path(config["user_data_dir"]) / "sentiment_btc.csv"
        if sentiment_path.exists():
            self._sentiment_df = pd.read_csv(
                sentiment_path,
                parse_dates=["date"],
            )
            self._sentiment_df["date"] = pd.to_datetime(
                self._sentiment_df["date"], utc=True
            )
            self._sentiment_df = self._sentiment_df.set_index("date")
        else:
            # 若 CSV 不存在,建立全中性的空 DataFrame
            self._sentiment_df = pd.DataFrame(columns=["sentiment_score"])

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # 技術指標
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["ema20"] = ta.EMA(dataframe, timeperiod=20)
        dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50)

        # 合併情緒分數:把 dataframe 的 date 欄位對齊到整點小時
        dataframe["date_hour"] = dataframe["date"].dt.floor("h")
        dataframe = dataframe.merge(
            self._sentiment_df[["sentiment_score"]],
            how="left",
            left_on="date_hour",
            right_index=True,
        )
        # 沒有情緒資料的時間點填 0(中性),策略照常執行
        dataframe["sentiment_score"] = dataframe["sentiment_score"].fillna(0.0)
        dataframe.drop(columns=["date_hour"], inplace=True)

        return dataframe

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        dataframe.loc[
            (
                # 技術條件:EMA 多頭排列 + RSI 低檔回升
                (dataframe["ema20"] > dataframe["ema50"]) &
                (dataframe["rsi"] < 45) &
                (dataframe["rsi"].shift(1) < dataframe["rsi"]) &
                # 情緒條件:過濾極度負面情緒環境
                (dataframe["sentiment_score"] >= self.sentiment_threshold) &
                (dataframe["volume"] > 0)
            ),
            "enter_long"
        ] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        dataframe.loc[
            (
                (dataframe["rsi"] > 70) |
                (dataframe["ema20"] < dataframe["ema50"])
            ),
            "exit_long"
        ] = 1
        return dataframe

為回測補充歷史情緒資料

回測需要歷史情緒資料,但你的 script 只會從現在開始累積。解決方法:用 LLM 「重跑」歷史新聞(CryptoCompare 提供有限的歷史新聞 API),或是用另一個簡化方法——先用隨機漫步合成假的歷史情緒資料來驗證架構,等累積了真實資料再做嚴肅回測:

# generate_synthetic_sentiment.py
# 目的:生成假的歷史情緒資料,只用來驗證架構正確性
# 注意:這個資料沒有真實預測意義,不能用來評估策略好壞

import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

start = datetime(2024, 1, 1, tzinfo=timezone.utc)
end = datetime(2025, 1, 1, tzinfo=timezone.utc)
hours = int((end - start).total_seconds() / 3600)

np.random.seed(42)
# 用平滑隨機漫步模擬情緒自相關性
raw = np.random.randn(hours) * 0.3
smoothed = pd.Series(raw).ewm(span=12).mean().clip(-1, 1)

df = pd.DataFrame({
    "date": [start + timedelta(hours=i) for i in range(hours)],
    "sentiment_score": smoothed.values.round(4),
    "news_count": np.random.randint(5, 25, size=hours),
})
df.to_csv("user_data/sentiment_btc.csv", index=False)
print(f"生成 {hours} 筆合成情緒資料")

跑對比回測:有情緒 vs 無情緒

同一段時間跑兩次回測,比較加入情緒過濾器前後的差異:

# 回測 1:有情緒過濾器
freqtrade backtesting \
  --config config.json \
  --strategy SentimentStrategy \
  --timerange 20240101-20250101 \
  --fees 0.001

# 把結果存下來(freqtrade 回測結果存在 user_data/backtest_results/)
cp user_data/backtest_results/*.json user_data/backtest_results/with_sentiment.json

接著把策略的情緒條件暫時關掉,做控制組:

# 在 populate_entry_trend 裡,暫時把情緒條件改成永遠為真:
# (dataframe["sentiment_score"] >= self.sentiment_threshold)
# 改成:
# (dataframe["sentiment_score"] >= -9999)  # 永遠通過

# 回測 2:無情緒過濾(控制組)
freqtrade backtesting \
  --config config.json \
  --strategy SentimentStrategy \
  --timerange 20240101-20250101 \
  --fees 0.001

用 freqtrade 內建的比較工具看差異:

freqtrade backtesting-analysis \
  --config config.json \
  --export-filename user_data/backtest_results/with_sentiment.json

解讀對比結果——別輕易相信 AI 訊號

對比回測的解讀有幾個關鍵問題要問:

== 有情緒過濾 ==
Total Profit:      +6.2%
Max Drawdown:      -8.4%
Trades:            61
Sharpe Ratio:      0.61

== 無情緒過濾(控制組) ==
Total Profit:      +4.1%
Max Drawdown:     -12.4%
Sharpe Ratio:      0.42

看起來有情緒過濾表現更好——但先別急著慶祝,問自己這三個問題:

1. 情緒資料是真實的還是合成的? 合成資料的回測只能驗證「架構能跑」,無法驗證「AI 情緒真的有預測力」。

2. 有沒有前視偏差? 如果你的情緒分數是「事後用已知新聞算的」,而不是「當下只用那根 K 線之前的新聞算的」,你等於讓策略偷看未來,回測完全失效。

3. 改善是因為情緒,還是因為少交易? 情緒過濾器讓交易次數從 87 減少到 61。驗證方法:把情緒閾值換成「隨機略過 30% 訊號」的規則,如果結果差不多,代表情緒評分本身沒有貢獻。