#!/usr/bin/env python3 """ 直接 API 晨報生成（替代 GitHub Copilot） 使用 DIKW 分層：Haiku → Sonnet → Opus """ import os import json import requests import yaml from datetime import datetime # DIKW 模型配置 DIKW_MODELS = { 'data': 'claude-3-5-haiku-20241022', # $0.80/M - 數據收集 'info': 'claude-3-5-sonnet-20241022', # $3/M - 信息整理 'knowledge': 'claude-3-5-sonnet-20241022', # $3/M - 知識分析 'wisdom': 'claude-3-opus-20240229' # $15/M - 智慧推演（僅用戶選3時） } class DirectAPIBriefing: def __init__(self): self.api_key = os.environ.get('ANTHROPIC_API_KEY') if not self.api_key: raise ValueError("需要設定 ANTHROPIC_API_KEY") self.base_url = "https://api.anthropic.com/v1/messages" self.headers = { "Content-Type": "application/json", "x-api-key": self.api_key, "anthropic-version": "2023-06-01" } def call_claude(self, model, prompt, max_tokens=4000): """直接調用 Claude API""" data = { "model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}] } try: response = requests.post(self.base_url, headers=self.headers, json=data) response.raise_for_status() return response.json()['content'][0]['text'] except Exception as e: print(f"API 調用錯誤: {e}") return None def fetch_news_data(self): """Layer D - 數據收集 (Haiku)""" print("🔍 Layer D - 數據收集...") stocks = os.environ.get('STOCKS', '').split(',') news_data = {} # 模擬新聞數據（實際應該爬蟲） for stock in stocks[:5]: # 限制前5檔測試 if stock.strip(): # 這裡應該實際爬蟲，現在用模擬數據 mock_news = f"{stock} 相關新聞：股價波動，市場關注..." # 用 Haiku 分析新聞 prompt = f""" 分析以下台股新聞，提取關鍵資訊： 股票代號：{stock} 新聞：{mock_news} 請以 JSON 格式回覆： {{ "stock": "{stock}", "sentiment": "positive/neutral/negative", "impact": "high/medium/low", "events": ["事件1", "事件2"], "key_points": ["要點1", "要點2"] }} """ result = self.call_claude(DIKW_MODELS['data'], prompt, 1000) if result: try: news_data[stock] = json.loads(result) except: news_data[stock] = {"error": "解析失敗", "raw": result} return news_data def analyze_information(self, news_data): """Layer I+K - 信息知識分析 (Sonnet)""" print("🧠 Layer I+K - 信息知識分析...") # 載入評分卡數據 watchlist_data = self.load_watchlist() prompt = f""" 作為台股投研分析師，請根據 DIKW 架構分析以下數據： 新聞數據：{json.dumps(news_data, ensure_ascii=False, indent=2)} 評分卡數據：{json.dumps(watchlist_data, ensure_ascii=False, indent=2)[:1000]}... 請參考專業投研報告標準生成今日晨報： # 台股投研晨報 {datetime.now().strftime('%Y-%m-%d')} ## 🚨 關鍵市場背景：宏觀脈絡分析 ### 國際政經動態 - 美股收盤影響：(分析隔夜美股走勢對台股的預期影響) - 重大政策變化：(如Fed決議、地緣政治、貿易政策等) ### 台股預期影響 - 開盤預估：跳空高開/平開/跳空低開 - 資金流向：風險性資產 vs 避險資產 - 情緒指標：Risk On vs Risk Off ## 📊 變動優先 Top 5：技術面+籌碼面整合 ### 🚀 強力反攻組 (技術超賣+法人轉向) 1) 代號 公司｜產業｜總分 XX (等級)｜Δ +X - 技術指標：K值 XX (低檔轉折) / RSI XX (脫離超賣) - 籌碼動向：外資轉買/投信加碼/自營避險 - 操盤策略：[市價追進/回測加碼/分批建倉] - 目標價位：短線看 XX 元 ### ⚠️ 高檔震盪組 (指標過熱+背離警示) 2) 代號 公司｜產業｜總分 XX (等級)｜Δ +/-X - 技術指標：K值 XX (高檔背離) / RSI XX (過熱區) - 風險提醒：獲利了結壓力/追高風險 ### 📉 避險撤退組 (資金排擠效應) 3) 代號 公司｜產業｜避險題材消失 - 撤退理由：風險偏好回升、避險需求降低 ## 📈 今日操盤戰略 ### 開盤策略 (09:00-09:15) - **大膽搶進**：(列出最具反彈潛力的2-3檔) - **觀察名單**：(技術面待確認的標的) - **避開地雷**：(高檔風險股) ### 盤中策略 & 尾盤策略 - 輪動邏輯、加減碼時點、週五效應考量 ## 🎯 風險控制 - 關鍵支撐壓力位 - 停損停利建議 --- **回覆 1/2/3 獲得進階分析：** 1 - 個股深度建議 (具體進出點位) 2 - 產業輪動分析 (資金流向邏輯) 3 - 情境推演框架 (樂觀/中性/悲觀三情境) 請以專業投研機構的標準，結合宏觀視野、技術分析、籌碼判讀，產出可操作的投資建議。 """ result = self.call_claude(DIKW_MODELS['info'], prompt, 3000) return result def load_watchlist(self): """載入觀察名單評分""" watchlist = {} try: watchlist_dir = 'data/watchlist' if os.path.exists(watchlist_dir): for filename in os.listdir(watchlist_dir): if filename.endswith('.yml'): filepath = os.path.join(watchlist_dir, filename) with open(filepath, 'r', encoding='utf-8') as f: data = yaml.safe_load(f) watchlist[data['ticker']] = { 'name': data['name'], 'industry': data['industry'], 'scores': data['scores'], 'grade': data['scores'].get('grade', 'B') } except Exception as e: print(f"載入評分卡錯誤: {e}") return watchlist def generate_wisdom_analysis(self, briefing, user_choice): """Layer W - 智慧推演 (Opus，僅在需要時)""" if user_choice != "3": return None print("💡 Layer W - 智慧推演...") prompt = f""" 基於以下晨報，設計深度投資推演框架： {briefing} 請用「情境分析法」設計 3 個投資情境： ## 情境推演框架 ### 樂觀情境 (機率 30%) - 假設條件： - 股價預期： - 操作策略： ### 中性情境 (機率 50%) - 假設條件： - 股價預期： - 操作策略： ### 悲觀情境 (機率 20%) - 假設條件： - 股價預期： - 操作策略： ### 風險控制 - 停損點： - 部位控制： - 監控指標： 請提供具體可執行的投資思考框架。 """ result = self.call_claude(DIKW_MODELS['wisdom'], prompt, 2000) return result def save_results(self, briefing): """儲存結果""" today = datetime.now().strftime('%Y-%m-%d') # 儲存到 output/ os.makedirs('output', exist_ok=True) with open('output/briefing.md', 'w', encoding='utf-8') as f: f.write(briefing) # 同時儲存到 data/briefings/ os.makedirs('data/briefings', exist_ok=True) with open(f'data/briefings/briefing_{today}.md', 'w', encoding='utf-8') as f: f.write(briefing) print(f"✅ 晨報已儲存: briefing_{today}.md") def run_daily_briefing(self): """執行每日晨報生成流程""" print("🦉 台股投研晨報生成開始...") print("使用直接 API 調用 (DIKW 分層)") try: # Layer D - 數據收集 news_data = self.fetch_news_data() # Layer I+K - 信息知識分析 briefing = self.analyze_information(news_data) if briefing: # 儲存結果 self.save_results(briefing) print("✅ 晨報生成完成") return briefing else: print("❌ 晨報生成失敗") return None except Exception as e: print(f"❌ 執行錯誤: {e}") return None def main(): """主程序""" try: generator = DirectAPIBriefing() briefing = generator.run_daily_briefing() if briefing: print("\n" + "="*50) print("📊 成本估算 (本次調用)") print("="*50) print("Layer D (Haiku): ~2k tokens × $0.80/M = $0.002") print("Layer I+K (Sonnet): ~8k tokens × $3/M = $0.024") print("預估本次成本: ~$0.026") print("月預估成本 (30天): ~$0.78") print("vs GitHub Copilot $10/月: 節省 92%!") except Exception as e: print(f"主程序錯誤: {e}") if __name__ == "__main__": main()