468 lines
15 KiB
Python
468 lines
15 KiB
Python
#!/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 線圖表
|
|
|
|
})
|
|
|
|
---
|
|
|
|
## ⚠️ 風險提示
|
|
|
|
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()
|