This commit is contained in:
dsyoon
2025-08-07 12:35:12 +09:00
parent edf8185d8c
commit f861487107
2 changed files with 155 additions and 117 deletions

View File

@@ -87,9 +87,9 @@ def buy_ticker(symbol, data):
BUY_AMOUNT = 6000
if data['buy_signal'].iloc[-1] == 'movingaverage':
BUY_AMOUNT = 50000
BUY_AMOUNT = 70000
elif data['buy_signal'].iloc[-1] == 'deviation40':
BUY_AMOUNT = 7000
BUY_AMOUNT = 40000
elif data['buy_signal'].iloc[-1] == 'deviation240':
BUY_AMOUNT = 6000

View File

@@ -1,5 +1,3 @@
import time
from datetime import datetime
from dateutil.relativedelta import relativedelta
import pandas as pd
import yfinance as yf
@@ -68,9 +66,10 @@ def fetch_price_history(symbol: str, interval_minutes: int, days: int = 30) -> p
def analyze_bottom_period(symbol: str, interval_minutes: int, days: int = 90):
"""저점 기간(6월 22일~7월 9일) 분석"""
"""저점 기간(6월 22일~7월 9일) 분석 - 최적화된 버전"""
data = fetch_price_history(symbol, interval_minutes, days)
data = calculate_technical_indicators(data)
data = check_buy_point(data, simulation=True) # 한 번만 계산
print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}")
print(f"총 데이터 수: {len(data)}")
@@ -83,7 +82,7 @@ def analyze_bottom_period(symbol: str, interval_minutes: int, days: int = 90):
if len(bottom_data) == 0:
print("저점 기간 데이터가 없습니다.")
return
return None, []
print(f"\n저점 기간 데이터: {bottom_data.index[0]} ~ {bottom_data.index[-1]}")
print(f"저점 기간 데이터 수: {len(bottom_data)}")
@@ -125,32 +124,14 @@ def analyze_bottom_period(symbol: str, interval_minutes: int, days: int = 90):
print(f"\n실제 저점:")
print(f"날짜: {actual_bottom_date}")
print(f"가격: {actual_bottom_price:.4f}")
# 실제 저점에서 RSI 출력 제거
# print(f"RSI: {bottom_data.loc[actual_bottom_idx, 'RSI']:.2f}")
print(f"볼린저 하단 대비: {((actual_bottom_price - bottom_data.loc[actual_bottom_idx, 'Lower']) / bottom_data.loc[actual_bottom_idx, 'Lower'] * 100):.2f}%")
# 6. 매수 신호 분석
# 6. 매수 신호 분석 - 최적화된 버전
print(f"\n=== 매수 신호 분석 ===")
# 현재 매수 조건으로 저점 기간에서 매수 신호가 몇 개 발생하는지 확인
alerts = []
debug_info = []
for i in range(len(bottom_data)):
slice_df = data.iloc[:data.index.get_loc(bottom_data.index[i]) + 1]
slice_df = check_buy_point(slice_df, simulation=True)
info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes)
if info:
debug_info.append({
'date': bottom_data.index[i],
'price': bottom_data['Close'].iloc[i],
'alert': info['alert'],
'details': info['details']
})
if info['alert']:
alerts.append((bottom_data.index[i], bottom_data['Close'].iloc[i]))
# 이미 계산된 매수 포인트에서 저점 기간만 필터링
bottom_alerts = bottom_data[bottom_data['buy_point'] == 1]
alerts = [(idx, row['Close']) for idx, row in bottom_alerts.iterrows()]
print(f"저점 기간 매수 신호 수: {len(alerts)}")
@@ -169,25 +150,22 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}")
print(f"총 데이터 수: {len(data)}")
# 파라미터 후보군 (표준화된 기술적 분석 기준)
param_candidates = [
{'rsi_oversold': 32, 'bb_distance': 0.06, 'near_low_tolerance': 0.01, 'volume_multiplier': 2.0}, # 기본 설정
{'rsi_oversold': 30, 'bb_distance': 0.05, 'near_low_tolerance': 0.008, 'volume_multiplier': 2.5}, # 엄격한 설정
{'rsi_oversold': 28, 'bb_distance': 0.04, 'near_low_tolerance': 0.005, 'volume_multiplier': 3.0}, # 매우 엄격한 설정
]
# check_buy_point에서 이미 계산된 매수 포인트를 사용
alerts = []
for params in param_candidates:
alerts.clear()
for i in range(len(data)):
slice_df = data.iloc[: i + 1]
slice_df = check_buy_point(slice_df, simulation=True)
info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes, params=params)
if info and info['alert']:
alerts.append((slice_df.index[-1], slice_df['Close'].iloc[-1]))
for i in range(len(data)):
if data['buy_point'].iloc[i] == 1:
alerts.append((data.index[i], data['Close'].iloc[i]))
print(f"\n총 매수 신호 수: {len(alerts)}")
# 매수 신호 유형별 통계 출력
ma_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'movingaverage')])
dev40_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation40')])
dev240_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation240')])
print(f" - MA 신호: {ma_signals}")
print(f" - Dev40 신호: {dev40_signals}")
print(f" - Dev240 신호: {dev240_signals}")
# 서브플롯 생성 (가격 + Deviation)
fig, (ax1, ax2) = plt.subplots(2, 1, sharex=True, figsize=(15, 8), height_ratios=[3, 1])
@@ -228,8 +206,39 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
fig.canvas.mpl_connect('scroll_event', on_scroll)
# 캔들스틱 차트 추가 (matplotlib 기본 기능 사용)
import matplotlib.dates as mdates
# 캔들스틱 데이터 준비
ohlc_data = []
for i, (idx, row) in enumerate(data.iterrows()):
ohlc_data.append([
mdates.date2num(idx),
row['Open'],
row['High'],
row['Low'],
row['Close']
])
# 캔들스틱 그리기 (matplotlib 기본 기능으로 구현)
for ohlc in ohlc_data:
date, open_price, high, low, close = ohlc
# 캔들 색상 결정
color = 'red' if close >= open_price else 'blue'
# 캔들 몸통 그리기
body_height = abs(close - open_price)
body_bottom = min(open_price, close)
rect = plt.Rectangle((date - 0.3, body_bottom), 0.6, body_height,
facecolor=color, edgecolor='black', alpha=0.7)
ax1.add_patch(rect)
# 캔들 심지 그리기
ax1.plot([date, date], [low, high], color='black', linewidth=1)
# 메인 차트 (가격, 이동평균선, 볼린저 밴드)
line_close = ax1.plot(data.index, data["Close"], label="종가", color="black", linewidth=1.5)[0]
line_close = ax1.plot(data.index, data["Close"], label="종가", color="black", linewidth=1.5, alpha=0.8)[0]
line_ma5 = ax1.plot(data.index, data["MA5"], label="MA5", color="red", linewidth=1)[0]
line_ma20 = ax1.plot(data.index, data["MA20"], label="MA20", color="blue", linewidth=1)[0]
line_ma40 = ax1.plot(data.index, data["MA40"], label="MA40", color="green", linewidth=1)[0]
@@ -245,55 +254,39 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
ax1.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1)
# 매수 신호 표시
scatter_buy = None
if alerts:
times, prices = zip(*alerts)
scatter_buy = ax1.scatter(times, prices, facecolors='none', edgecolors='red', linewidths=2, s=150, zorder=6, label='매수신호')
for time in times:
ax1.axvline(x=time, color='red', linestyle='--', alpha=0.3)
# 매수 포인트 탐지 및 표시
# 'buy_point' 열 추가
data['buy_point'] = 0
data['buy_signal'] = ''
for i in range(1, len(data)):
# 이동평균선 기반 매수 조건
if all(data[f'MA{n}'].iloc[i] < data['MA720'].iloc[i] for n in [5, 20, 40, 120, 200, 240]) and \
all(data[f'MA{n}'].iloc[i] > data[f'MA{n}'].iloc[i-1] for n in [5, 20, 40, 120, 200, 240]) and \
data['MA720'].iloc[i] < data['MA1440'].iloc[i]:
data.at[data.index[i], 'buy_signal'] = 'movingaverage'
data.at[data.index[i], 'buy_point'] = 1
# Deviation40(이격도 40) 기반 매수 조건: 90 이하에서 상승 전환
if data['Deviation40'].iloc[i - 1] < data['Deviation40'].iloc[i] and data['Deviation40'].iloc[i - 1] <= 90:
data.at[data.index[i], 'buy_signal'] = 'deviation40'
data.at[data.index[i], 'buy_point'] = 1
# Deviation240(이격도 240) 기반 매수 조건: 90 이하에서 상승 전환
if data['Deviation240'].iloc[i - 1] < data['Deviation240'].iloc[i] and data['Deviation240'].iloc[i - 1] <= 90:
data.at[data.index[i], 'buy_signal'] = 'deviation240'
data.at[data.index[i], 'buy_point'] = 1
# 매수 포인트를 신호 유형별로 다르게 표시
# 이동평균선 기반 매수 포인트 (빨간 동그라미)
# 매수 포인트를 신호 유형별로 다르게 표시 (수직선 포함)
# 이동평균선 기반 매수 포인트
ma_buy_points = data[(data['buy_point'] == 1) & (data['buy_signal'] == 'movingaverage')]
scatter_ma_buy_points = ax1.scatter(ma_buy_points.index, ma_buy_points['Close'], color='red', s=100, zorder=5, label='MA 매수 포인트')
scatter_ma_buy_points = None
if len(ma_buy_points) > 0:
scatter_ma_buy_points = ax1.scatter(ma_buy_points.index, ma_buy_points['Close'], color='red', s=150, zorder=10, label='MA 매수 포인트', marker='o')
for time in ma_buy_points.index:
ax1.axvline(x=time, color='red', linestyle='-', alpha=0.5, linewidth=1)
# Deviation40 기반 매수 포인트 (속이 빈 빨간 점선 원)
# Deviation40 기반 매수 포인트
dev40_buy_points = data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation40')]
scatter_dev40_buy_points = ax1.scatter(dev40_buy_points.index, dev40_buy_points['Close'],
facecolors='none', edgecolors='red', linestyle='--',
linewidth=1, s=150, zorder=5, label='Dev40 매수 포인트')
scatter_dev40_buy_points = None
if len(dev40_buy_points) > 0:
scatter_dev40_buy_points = ax1.scatter(dev40_buy_points.index, dev40_buy_points['Close'],
facecolors='none', edgecolors='red', linestyle='--',
linewidth=2, s=200, zorder=10, label='Dev40 매수 포인트')
for time in dev40_buy_points.index:
ax1.axvline(x=time, color='red', linestyle='--', alpha=0.5, linewidth=1)
# Deviation240 기반 매수 포인트 (속이 빈 파란 점선 원)
# Deviation240 기반 매수 포인트
dev240_buy_points = data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation240')]
scatter_dev240_buy_points = ax1.scatter(dev240_buy_points.index, dev240_buy_points['Close'],
facecolors='none', edgecolors='blue', linestyle='--',
linewidth=1, s=150, zorder=5, label='Dev240 매수 포인트')
scatter_dev240_buy_points = None
if len(dev240_buy_points) > 0:
scatter_dev240_buy_points = ax1.scatter(dev240_buy_points.index, dev240_buy_points['Close'],
facecolors='none', edgecolors='blue', linestyle='--',
linewidth=2, s=200, zorder=10, label='Dev240 매수 포인트')
for time in dev240_buy_points.index:
ax1.axvline(x=time, color='blue', linestyle='--', alpha=0.5, linewidth=1)
# 마우스 오버 기능 추가 (이동평균선 매수 포인트)
if len(ma_buy_points) > 0:
if scatter_ma_buy_points is not None:
cursor = mplcursors.cursor(scatter_ma_buy_points, hover=True)
cursor.connect("add", lambda sel: sel.annotation.set_text(
f'MA 매수신호\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}'
@@ -301,7 +294,7 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
cursor.connect("remove", lambda sel: sel.annotation.set_visible(False))
# 마우스 오버 기능 추가 (Deviation40 매수 포인트)
if len(dev40_buy_points) > 0:
if scatter_dev40_buy_points is not None:
cursor_dev40 = mplcursors.cursor(scatter_dev40_buy_points, hover=True)
cursor_dev40.connect("add", lambda sel: sel.annotation.set_text(
f'Dev40 매수신호\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}'
@@ -309,20 +302,14 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
cursor_dev40.connect("remove", lambda sel: sel.annotation.set_visible(False))
# 마우스 오버 기능 추가 (Deviation240 매수 포인트)
if len(dev240_buy_points) > 0:
if scatter_dev240_buy_points is not None:
cursor_dev240 = mplcursors.cursor(scatter_dev240_buy_points, hover=True)
cursor_dev240.connect("add", lambda sel: sel.annotation.set_text(
f'Dev240 매수신호\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}'
))
cursor_dev240.connect("remove", lambda sel: sel.annotation.set_visible(False))
# 매수 신호에도 마우스 오버 기능 추가
if scatter_buy is not None:
cursor2 = mplcursors.cursor(scatter_buy, hover=True)
cursor2.connect("add", lambda sel: sel.annotation.set_text(
f'매수신호\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}'
))
cursor2.connect("remove", lambda sel: sel.annotation.set_visible(False))
# 모든 봉에 마우스 오버 기능 추가
cursor3 = mplcursors.cursor(line_close, hover=True)
@@ -331,10 +318,17 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
))
cursor3.connect("remove", lambda sel: sel.annotation.set_visible(False))
ax1.set_ylabel("가격")
ax1.set_ylabel("가격 (KRW)")
ax1.set_title(f"{symbol} 차트 - 캔들스틱 & 이동평균선", fontsize=14)
ax1.grid(True, linestyle='--', alpha=0.3)
# 홈 버튼 추가 (초기화 기능)
home_button = ax1.text(0.02, 0.98, '', transform=ax1.transAxes,
bbox=dict(boxstyle="round,pad=0.3", facecolor='lightblue', alpha=0.8),
fontsize=10, ha='left', va='top', picker=True)
# --- 범례 생성 및 인터랙티브 토글 ---
legend = ax1.legend(loc='upper left', fontsize=10)
ax1.grid(True, linestyle='--', alpha=0.5)
legend = ax1.legend(loc='upper left', bbox_to_anchor=(0.02, 0.85), fontsize=10)
# 범례 클릭 시 해당 선 토글 기능
lined = {}
@@ -349,10 +343,15 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
legend_handles = legend.get_lines()
plot_lines = [line_close, line_ma5, line_ma20, line_ma40, line_ma120,
line_ma200, line_ma240, line_ma720, line_ma1440,
line_upper, line_lower, scatter_ma_buy_points, scatter_dev40_buy_points, scatter_dev240_buy_points]
# 매수신호 scatter가 있으면 포함
if scatter_buy is not None:
plot_lines.append(scatter_buy)
line_upper, line_lower]
# None이 아닌 scatter plot만 추가
if scatter_ma_buy_points is not None:
plot_lines.append(scatter_ma_buy_points)
if scatter_dev40_buy_points is not None:
plot_lines.append(scatter_dev40_buy_points)
if scatter_dev240_buy_points is not None:
plot_lines.append(scatter_dev240_buy_points)
# zip 길이가 짧은 쪽에 맞춰 매핑
for leg_handle, orig in zip(legend_handles, plot_lines):
@@ -361,6 +360,19 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
def on_pick(event):
leg_handle = event.artist
# 홈 버튼 클릭 처리
if leg_handle == home_button:
# 그래프 초기화
ax1.set_xlim(matplotlib.dates.date2num(data.index[0]), matplotlib.dates.date2num(data.index[-1]))
ax1.set_ylim(data['Low'].min() * 0.99, data['High'].max() * 1.01)
ax2.set_xlim(ax1.get_xlim())
ax2.set_ylim(data[['Deviation5', 'Deviation20', 'Deviation40', 'Deviation120', 'Deviation200', 'Deviation240', 'Deviation720', 'Deviation1440']].min().min() * 0.95,
data[['Deviation5', 'Deviation20', 'Deviation40', 'Deviation120', 'Deviation200', 'Deviation240', 'Deviation720', 'Deviation1440']].max().max() * 1.05)
fig.canvas.draw_idle()
return
# 기존 범례 클릭 처리
orig = lined.get(leg_handle)
if orig is None:
return
@@ -372,23 +384,26 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
fig.canvas.mpl_connect('pick_event', on_pick)
# Deviation subplot
line_dev5 = ax2.plot(data.index, data['Deviation5'], color='red', label='Dev5')[0]
line_dev20 = ax2.plot(data.index, data['Deviation20'], color='blue', label='Dev20')[0]
line_dev40 = ax2.plot(data.index, data['Deviation40'], color='green', label='Dev40')[0]
line_dev120 = ax2.plot(data.index, data['Deviation120'], color='purple', label='Dev120')[0]
line_dev200 = ax2.plot(data.index, data['Deviation200'], color='brown', label='Dev200')[0]
line_dev240 = ax2.plot(data.index, data['Deviation240'], color='darkred', label='Dev240')[0]
line_dev720 = ax2.plot(data.index, data['Deviation720'], color='cyan', label='Dev720')[0]
line_dev1440 = ax2.plot(data.index, data['Deviation1440'], color='magenta', label='Dev1440')[0]
# Deviation subplot (이격도 보조지표)
line_dev5 = ax2.plot(data.index, data['Deviation5'], color='red', label='Dev5', linewidth=1)[0]
line_dev20 = ax2.plot(data.index, data['Deviation20'], color='blue', label='Dev20', linewidth=1)[0]
line_dev40 = ax2.plot(data.index, data['Deviation40'], color='green', label='Dev40', linewidth=2)[0]
line_dev120 = ax2.plot(data.index, data['Deviation120'], color='purple', label='Dev120', linewidth=1)[0]
line_dev200 = ax2.plot(data.index, data['Deviation200'], color='brown', label='Dev200', linewidth=1)[0]
line_dev240 = ax2.plot(data.index, data['Deviation240'], color='darkred', label='Dev240', linewidth=2)[0]
line_dev720 = ax2.plot(data.index, data['Deviation720'], color='cyan', label='Dev720', linewidth=1)[0]
line_dev1440 = ax2.plot(data.index, data['Deviation1440'], color='magenta', label='Dev1440', linewidth=1)[0]
cursor_dev = mplcursors.cursor([line_dev5, line_dev20, line_dev40, line_dev120, line_dev200, line_dev240, line_dev720, line_dev1440], hover=True)
cursor_dev.connect("add", lambda sel: sel.annotation.set_text(
f"{sel.artist.get_label()}\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime('%Y-%m-%d %H:%M')}\n값: {sel.target[1]:.2f}"
))
line_h90 = ax2.axhline(90, color='red', linestyle='--', linewidth=1, label='90')
line_h95 = ax2.axhline(95, color='green', linestyle='--', linewidth=1, label='93')
ax2.set_ylabel('Deviation %')
# 이격도 기준선 추가
line_h90 = ax2.axhline(90, color='red', linestyle='--', linewidth=2, label='90', alpha=0.7)
line_h95 = ax2.axhline(95, color='green', linestyle='--', linewidth=2, label='95', alpha=0.7)
line_h100 = ax2.axhline(100, color='black', linestyle='-', linewidth=1, label='100', alpha=0.5)
ax2.set_ylabel('이격도 (%)')
ax2.set_title('이격도 보조지표', fontsize=12)
legend2 = ax2.legend(loc='upper left', fontsize=9)
# Deviation subplot 범례 클릭 토글 기능
@@ -410,6 +425,9 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
ax2.grid(True, linestyle='--', alpha=0.3)
plt.tight_layout()
print("그래프를 표시합니다...")
print(f"매수 포인트 수: MA={len(ma_buy_points)}, Dev40={len(dev40_buy_points)}, Dev240={len(dev240_buy_points)}")
# -------- 확대/축소 및 이동 기능 --------
press = {}
@@ -466,15 +484,35 @@ if __name__ == "__main__":
target_coins = ['ADA','APE','ARB','BONK','HBAR','LINK','ONDO','PEPE','SEI','SHIB','STORJ','SUI','TON','TRX','WLD','XLM','XRP']
#target_coins = ['APE']
# 그래프 표시 여부 설정 (성능 향상을 위해 기본값은 False)
show_graphs = True # True로 설정하면 각 코인마다 그래프 표시
for symbol in target_coins:
print(f"\n=== {symbol} 저점 기간 분석 시작 ===")
try:
# 저점 기간 분석
bottom_data, alerts = analyze_bottom_period(symbol, interval, days)
# 전체 기간 시뮬레이션
# 전체 기간 시뮬레이션 (그래프 표시 옵션 추가)
print(f"\n=== {symbol} 전체 기간 시뮬레이션 ===")
run_simulation(symbol, interval, days)
if show_graphs:
run_simulation(symbol, interval, days)
else:
# 그래프 없이 빠른 분석만 수행
data = fetch_price_history(symbol, interval, days)
data = calculate_technical_indicators(data)
data = check_buy_point(data, simulation=True)
# 매수 신호 통계만 출력
total_buy_signals = len(data[data['buy_point'] == 1])
ma_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'movingaverage')])
dev40_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation40')])
dev240_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation240')])
print(f"총 매수 신호: {total_buy_signals}")
print(f" - MA 신호: {ma_signals}")
print(f" - Dev40 신호: {dev40_signals}")
print(f" - Dev240 신호: {dev240_signals}")
except Exception as e:
print(f"Error analyzing {symbol}: {str(e)}")