#!/usr/bin/env python3 """ Sakata Analyzer v2.0 - 酒田戰法分析器 (改良版) Main entry point for candlestick pattern analysis. 改良重點: 1. 整合結構分析 (頸線、量價背離) 2. 改進報告格式 (顯示型態階段、關鍵價位) 3. 趨勢背景信息 Usage: python sakata_analyzer.py --ticker AAPL python sakata_analyzer.py --ticker TSLA --days 180 """ import argparse import os import time from datetime import datetime, timedelta import pandas as pd import yfinance as yf from pattern_detector import detect_all_patterns, summarize_detected_patterns from chart_plotter import create_sakata_chart from signal_generator import SignalGenerator # Rate limiting settings for Yahoo Finance REQUEST_DELAY = 1.0 # seconds between requests MAX_RETRIES = 3 RETRY_DELAY = 5.0 # seconds def fetch_stock_data(ticker: str, days: int = 120, retries: int = 0) -> pd.DataFrame: """ Fetch OHLCV data from Yahoo Finance with rate limiting. """ print(f"📊 正在取得 {ticker} 的 {days} 日數據...") try: time.sleep(REQUEST_DELAY) end_date = datetime.now() start_date = end_date - timedelta(days=days + 30) stock = yf.Ticker(ticker) df = stock.history(start=start_date, end=end_date) if df.empty: raise ValueError(f"No data found for {ticker}") df = df.tail(days) print(f"✅ 成功取得 {len(df)} 筆數據 ({df.index[0].strftime('%Y-%m-%d')} ~ {df.index[-1].strftime('%Y-%m-%d')})") return df except Exception as e: if retries < MAX_RETRIES: print(f"⚠️ 請求失敗，{RETRY_DELAY} 秒後重試... ({retries + 1}/{MAX_RETRIES})") time.sleep(RETRY_DELAY * (retries + 1)) return fetch_stock_data(ticker, days, retries + 1) else: raise Exception(f"Failed to fetch data for {ticker} after {MAX_RETRIES} retries: {e}") def generate_structure_section(analysis: dict, pattern_type: str) -> str: """ 生成結構分析區塊 (頸線、量價背離)。 """ if not analysis.get('detected'): return "" name = '三山' if pattern_type == 'sanzan' else '三川' stage = analysis.get('stage', 0) section = f""" ## 🏔️ {name}型態結構分析 | 項目 | 數值 | |------|------| | **型態階段** | {stage}/5 {'⚠️ 預警中' if stage < 4 else '✅ 確立'} | """ # 頸線信息 neckline = analysis.get('neckline', {}) if neckline and neckline.get('neckline'): status = '✅ 已跌破' if neckline.get('confirmed') else '⏳ 未跌破' if pattern_type == 'sansen': status = '✅ 已突破' if neckline.get('confirmed') else '⏳ 未突破' section += f"""| **頸線價位** | ${neckline['neckline']:.2f} | | **頸線狀態** | {status} | | **距離頸線** | {neckline.get('distance_pct', 0):.1f}% | """ # 量價背離 vol_div = analysis.get('volume_divergence', {}) if vol_div and vol_div.get('divergence'): section += f"""| **量價背離** | ⚠️ {vol_div['description']} | """ # 關鍵價位 if pattern_type == 'sanzan' and analysis.get('peaks'): avg_peak = analysis.get('avg_peak', 0) section += f"""| **平均山頂** | ${avg_peak:.2f} | """ elif pattern_type == 'sansen' and analysis.get('troughs'): avg_trough = analysis.get('avg_trough', 0) section += f"""| **平均谷底** | ${avg_trough:.2f} | """ # 型態階段說明 stage_desc = { 1: '僅形成一個頂/底', 2: '形成兩個頂/底', 3: '第三頂/底形成中', 4: '型態成形，等待確認', 5: '型態確立，已突破/跌破頸線' } section += f""" ### 階段說明 {stage_desc.get(stage, '未知')} """ return section def generate_trend_section(trend_ctx: dict, current_price: float) -> str: """ 生成趨勢背景區塊。 """ position_map = {'high': '高檔', 'mid': '中間', 'low': '低檔'} trend_map = {'uptrend': '上升趨勢 📈', 'downtrend': '下降趨勢 📉', 'sideways': '盤整 ↔️'} ma20 = trend_ctx.get('ma20', [0])[-1] if len(trend_ctx.get('ma20', [])) > 0 else 0 ma60 = trend_ctx.get('ma60', [0])[-1] if len(trend_ctx.get('ma60', [])) > 0 else 0 section = f""" ## 📊 趨勢背景 | 指標 | 數值 | 說明 | |------|------|------| | **當前位置** | {position_map.get(trend_ctx.get('position', 'mid'), '中間')} | 近60日區間位置 ({trend_ctx.get('position_pct', 0)*100:.0f}%) | | **趨勢方向** | {trend_map.get(trend_ctx.get('trend', 'sideways'), '盤整')} | 基於 MA20 斜率 | | **MA20** | ${ma20:.2f} | {'股價在上方 ✅' if current_price > ma20 else '股價在下方 ❌'} | | **MA60** | ${ma60:.2f} | {'股價在上方 ✅' if current_price > ma60 else '股價在下方 ❌'} | | **近期高點** | ${trend_ctx.get('recent_high', 0):.2f} | | | **近期低點** | ${trend_ctx.get('recent_low', 0):.2f} | | """ return section def generate_markdown_report( ticker: str, df: pd.DataFrame, detected_patterns: list, signals: list, chart_path: str, analysis_result: dict, output_dir: str = './output' ) -> str: """ Generate an improved Markdown analysis report with structure analysis. """ signal_gen = SignalGenerator() summary = signal_gen.summarize_signals(signals) recommendation = signal_gen.get_latest_recommendation( signals, current_price=df.iloc[-1]['Close'], current_date=df.index[-1] ) current_price = df.iloc[-1]['Close'] trend_ctx = analysis_result.get('trend_context', {}) sanzan_analysis = analysis_result.get('sanzan_analysis', {}) sansen_analysis = analysis_result.get('sansen_analysis', {}) # 判斷是否有結構型態 has_structure = sanzan_analysis.get('detected') or sansen_analysis.get('detected') # Build report report = f"""# {ticker} 酒田戰法分析報告 v2.0 **生成時間**: {datetime.now().strftime('%Y-%m-%d %H:%M')} **數據範圍**: {df.index[0].strftime('%Y-%m-%d')} ~ {df.index[-1].strftime('%Y-%m-%d')} ({len(df)} 日) **當前價格**: ${current_price:.2f} --- ## 📊 核心結論 | 指標 | 數值 | |------|------| | 總信號數 | {summary['total']} | | 多頭信號 | {summary['bullish']} 🟢 | | 空頭信號 | {summary['bearish']} 🔴 | | 整體偏向 | **{summary['bias']}** | | 信心度 | {'⭐' * int(summary['avg_strength'])} ({summary['avg_strength']}/5) | """ # 添加趨勢背景 report += generate_trend_section(trend_ctx, current_price) # 添加結構分析 (如果有) if sanzan_analysis.get('detected'): report += generate_structure_section(sanzan_analysis, 'sanzan') if sansen_analysis.get('detected'): report += generate_structure_section(sansen_analysis, 'sansen') # 交易建議 report += """--- ## 🎯 最新交易建議 """ if recommendation: # v2.2: 根據狀態決定顯示方式 status = recommendation.get('status', 'ACTIVE') status_reason = recommendation.get('status_reason', '') days_diff = recommendation.get('days_since_signal', 0) # 決定 emoji 和建議文字 if recommendation['recommendation'] == 'HOLD': rec_emoji = '⏸️ 觀望' elif recommendation['recommendation'] == 'BUY': rec_emoji = '🟢 買入' else: rec_emoji = '🔴 賣出' # 狀態標籤 status_badge = '' if status == 'EXPIRED': status_badge = '⚠️ 已過期' elif status == 'STOPPED_OUT': status_badge = '🛑 已停損' elif status == 'SPECIAL': status_badge = '⚡ 特殊情況' elif status == 'ACTIVE': status_badge = '✅ 有效' report += f"""| 項目 | 數值 | |------|------| | **建議** | {rec_emoji} | | **信號狀態** | {status_badge} | | **觸發型態** | {recommendation['pattern']} | | **信號日期** | {recommendation['date'].strftime('%Y-%m-%d') if hasattr(recommendation['date'], 'strftime') else recommendation['date']} ({days_diff} 天前) | | **建議進場** | ${recommendation['entry']:.2f} | | **停損價位** | ${recommendation['stop_loss']:.2f} | | **目標價位** | ${recommendation['target']:.2f} | | **信號強度** | {recommendation['strength']} | """ # 狀態原因說明 if status_reason: report += f"""> **狀態說明**: {status_reason} """ # 連續窗口警示 window_status = recommendation.get('window_status', {}) if window_status.get('warning'): report += f"""> **酒田特殊警示**: {window_status['warning']} """ # 只有 ACTIVE 狀態才顯示盈虧比 if status == 'ACTIVE': risk = abs(recommendation['entry'] - recommendation['stop_loss']) reward = abs(recommendation['target'] - recommendation['entry']) rr_ratio = reward / risk if risk > 0 else 0 report += f"""**盈虧比 (R/R)**: 1 : {rr_ratio:.1f} """ else: report += "> ⚠️ 目前無明確交易信號，建議觀望。\n\n" # 偵測到的型態 (去重後顯示) report += """--- ## 📈 偵測到的型態 | 日期 | 型態 | 英文名稱 | 方向 | 強度 | |------|------|---------|------|------| """ # 去重: 同一天同一型態只顯示一次 seen = set() unique_patterns = [] for pattern in detected_patterns[:20]: key = (pattern['date'].strftime('%Y-%m-%d') if hasattr(pattern['date'], 'strftime') else str(pattern['date'])[:10], pattern['name']) if key not in seen: seen.add(key) unique_patterns.append(pattern) for pattern in unique_patterns[:12]: date_str = pattern['date'].strftime('%m/%d') if hasattr(pattern['date'], 'strftime') else str(pattern['date'])[:10] direction = '🟢 多頭' if pattern['direction'] == 'bullish' else '🔴 空頭' strength = '⭐' * min(5, max(1, pattern['strength'] // 20)) report += f"| {date_str} | {pattern['name']} | {pattern['english']} | {direction} | {strength} |\n" report += f""" --- ## 📉 K 線圖表 ![{ticker} 酒田戰法分析]({os.path.basename(chart_path)}) --- ## ⚠️ 風險提示 1. 本分析僅供參考，不構成投資建議 2. K 線型態需配合成交量、趨勢等其他指標確認 3. **三山/三川型態需跌破/突破頸線才算確立** 4. 請嚴格執行停損紀律 5. 過去表現不代表未來結果 --- *由酒田戰法 Agent Skill v2.0 自動生成* """ # Save report os.makedirs(output_dir, exist_ok=True) report_path = os.path.join(output_dir, f'{ticker}_sakata.md') with open(report_path, 'w', encoding='utf-8') as f: f.write(report) return report_path def analyze(ticker: str, days: int = 120, output_dir: str = './output'): """ Run full Sakata analysis on a stock. """ print(f"\n{'='*60}") print(f"🏯 酒田戰法分析器 v2.0 - {ticker}") print(f"{'='*60}\n") # 1. Fetch data df = fetch_stock_data(ticker, days) # 2. Detect patterns (v2.0 - returns rich analysis) print("\n🔍 正在偵測 K 線型態 (含結構分析)...") analysis_result = detect_all_patterns(df) pattern_results = analysis_result['patterns'] trend_ctx = analysis_result['trend_context'] sanzan_analysis = analysis_result['sanzan_analysis'] sansen_analysis = analysis_result['sansen_analysis'] detected_patterns = summarize_detected_patterns(df, pattern_results) print(f"✅ 偵測到 {len(detected_patterns)} 個型態") # 顯示結構分析結果 if sanzan_analysis.get('detected'): print(f" 📉 三山型態: 階段 {sanzan_analysis.get('stage')}/5") if sansen_analysis.get('detected'): print(f" 📈 三川型態: 階段 {sansen_analysis.get('stage')}/5") # 3. Generate signals print("\n📈 正在產生交易信號...") signal_gen = SignalGenerator() signals = signal_gen.generate_signals(df, detected_patterns, trend_ctx) print(f"✅ 產生 {len(signals)} 個交易信號") # 4. Create chart print("\n🎨 正在繪製圖表...") chart_path = create_sakata_chart(df, detected_patterns, signals, ticker, output_dir) print(f"✅ 圖表已儲存: {chart_path}") # 5. Generate report print("\n📝 正在產生報告...") report_path = generate_markdown_report( ticker, df, detected_patterns, signals, chart_path, analysis_result, output_dir ) print(f"✅ 報告已儲存: {report_path}") # 6. Summary summary = signal_gen.summarize_signals(signals) recommendation = signal_gen.get_latest_recommendation( signals, current_price=df.iloc[-1]['Close'], current_date=df.index[-1] ) print(f"\n{'='*60}") print("📊 分析完成！") print(f"{'='*60}") print(f" 趨勢位置: {trend_ctx.get('position', 'mid')} | 趨勢方向: {trend_ctx.get('trend', 'sideways')}") print(f" 多頭信號: {summary['bullish']} | 空頭信號: {summary['bearish']}") print(f" 整體偏向: {summary['bias']}") if recommendation: print(f"\n 🎯 最新建議: {recommendation['recommendation']}") print(f" 型態: {recommendation['pattern']}") print(f" 進場: ${recommendation['entry']:.2f}") print(f" 停損: ${recommendation['stop_loss']:.2f}") print(f" 目標: ${recommendation['target']:.2f}") print(f"\n 📁 輸出檔案:") print(f" {chart_path}") print(f" {report_path}") print(f"{'='*60}\n") return { 'chart': chart_path, 'report': report_path, 'patterns': detected_patterns, 'signals': signals, 'summary': summary, 'analysis': analysis_result } def main(): parser = argparse.ArgumentParser( description='酒田戰法分析器 v2.0 - Sakata Candlestick Pattern Analyzer', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 範例: python sakata_analyzer.py --ticker AAPL python sakata_analyzer.py --ticker TSLA --days 180 python sakata_analyzer.py --ticker NVDA --output ./charts """ ) parser.add_argument( '--ticker', '-t', type=str, required=True, help='股票代碼 (例: AAPL, TSLA, 2330.TW)' ) parser.add_argument( '--days', '-d', type=int, default=120, help='分析天數 (預設: 120)' ) parser.add_argument( '--output', '-o', type=str, default='./output', help='輸出目錄 (預設: ./output)' ) args = parser.parse_args() analyze( ticker=args.ticker.upper(), days=args.days, output_dir=args.output ) if __name__ == '__main__': main()