dsyoon/DeepCoin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
dsyoon
2025-08-06 14:39:43 +09:00
parent 872cf788dd
commit f7870381bf
6 changed files with 362 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
PROMPT.txt Normal file
View File

@@ -0,0 +1,15 @@
#1.
봉을 모두 1500개를 출력하고 있습니다.
전반적으로 그래프를 이해하세요.
그리고 시가, 종가, 고가, 저가, 거래량을 활용하여 많은 기술적 분석을 시도하세요.
그리고 최적의 저점에 매수를 할 수 있도록 탐색코드를 작성하세요.
모든 코인에 동일 조건을 적용할 수 있도록 활용하는 수치는 모두 정규화를 시킨 이후에 시도해야 합니다.
코드에 탐색을 반영하고 실행하세요, 그리고 매수 시점을 표기하세요. 그리고 최적이 아니라면 다시 실행하세요. 최적의 매수시점을 찾을 때까지 반복하세요.
#2.
전제 소스코들 살펴보세요. 데이터는 모든 코인에 동일 적용할 수 있도록 표준화가 되어야 합니다. 그리고 기술적 분석을 이용하여 매수 타이밍을 탐색해야 합니다.
혹시 기술적 분석이 아닌 날짜가 매수조건에 들어가 있다면 그러한 부분은 모두 제거해주세요.

42
README.md
View File

@@ -3,6 +3,11 @@
## 개요
`AssetMonitor`는 주식‧ETF 및 암호화폐 시장을 실시간으로 감시하여 Bollinger Band, RSI, MACD, 이동평균(Golden-Cross), 거래량 등을 종합 분석한 **매수 후보(signals)**를 텔레그램으로 통보하는 자동화 봇입니다.
**주요 개선사항:**
- **데이터 표준화**: 모든 코인에 동일한 기술적 분석 기준 적용
- **순수 기술적 분석**: 날짜 기반 조건 제거, 기술적 지표만 사용
- **강화된 기술적 지표**: 스토캐스틱, MFI, OBV, ATR 등 추가 지표 활용
---
## 주요 구성 파일
@@ -20,20 +25,24 @@
2. **데이터 획득**
*주식 / ETF*: `FinanceDataReader`
*암호화폐*: 빗썸 **240분 봉** Open API
3. **기술적 지표 계산** (`calculate_technical_indicators`)
3. **데이터 표준화** (`normalize_data`)
- 모든 코인에 동일한 정규화 적용
- 20일 롤링 윈도우 기반 Min-Max 정규화
4. **기술적 지표 계산** (`calculate_technical_indicators`)
- Bollinger Band (기간 20, ±2σ)
- RSI(14)
- MACD(12-26-9)
- 단/중/장기 이동평균선(MA5/20/60)
- 거래량 MA5
4. **매수 후보 판정** (`check_buy_signals`)
- *아래 새로운 “매수 후보 전략” 섹션 참조*
5. **알림 발송** (`send_*_telegram_message`)
- **추가 지표**: 스토캐스틱, OBV, ATR, MFI
5. **매수 후보 판정** (`check_buy_signals`)
- *아래 새로운 "매수 후보 전략" 섹션 참조*
6. **알림 발송** (`send_*_telegram_message`)
multiprocessing Pool을 이용해 다중 메시지를 병렬로 전송합니다.
---
## 매수 후보 전략
## 매수 후보 전략 (표준화된 기술적 분석)
| 신호 | 변수명 | 조건 | 의미 |
|------|--------|------|------|
@@ -43,21 +52,26 @@
| 이동평균 골든크로스 | `ma_signal` | `이전 MA5 < 이전 MA20` **AND** `현재 MA5 ≥ 현재 MA20` | 단기 추세 ↗ 전환 |
| 거래량 급증 | `volume_signal` | `현재 거래량 > MA5 Volume × 1.5` | 수급 증가 |
| **U자 반등 돌파** | `breakout_signal` | ① 최근 `BREAKOUT_LOOKBACK`(30)개 캔들 동안 최고·최저가 차이가 `BUY_THRESHOLD`(15 %) 이상 하락 → ② **현재가가 그 최고가 돌파** | 하락 후 반등의 추세 전환 확인 |
| **장기 저항 돌파** | `long_breakout_signal` | 장기간 저항선 돌파 감지 | 장기 추세 전환 |
| **스토캐스틱 과매도** | `stoch_signal` | `%K < 20 AND %K > 이전 %K` | 스토캐스틱 과매도 반등 |
| **MFI 과매도** | `mfi_signal` | `MFI < 20 AND MFI > 이전 MFI` | 자금 흐름 과매도 반등 |
| **OBV 상승** | `obv_signal` | `현재 OBV > 이전 OBV × 1.1` | 거래량 가중 상승 |
| **ATR 급증** | `atr_signal` | `현재 ATR > ATR 20일 평균 × 1.5` | 변동성 급증 |
### 최종 매수 후보 결정 로직
```text
if breakout_signal:
buy = True # U자 반등 돌파 단독으로도 매수 후보
if breakout_signal or long_breakout_signal:
buy = True # 돌파 신호 단독으로도 매수 후보
else:
# ① 볼린저 + RSI 동시, 또는 ② (신호 ≥ 2개) & (볼린저 또는 RSI 포함)
buy = (bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))
# ① 볼린저 + RSI 동시, 또는 ② (신호 ≥ 3개) & (볼린저 또는 RSI 포함)
buy = (bb_signal and rsi_signal) or (signal_count >= 3 and (bb_signal or rsi_signal))
```
*`signal_count` = 위 6개 신호 중 True 개수*
*`signal_count` = 위 11개 신호 중 True 개수*
### 메시지 구성
- `🛒` : 최종 `buy=True`일 때 메시지 맨 앞에 부착
- `📊신호(n):` 뒤에 활성화된 신호 목록
- 볼린저/RSI/MACD/MA/거래량/Breakout 각각 표시
- `매수` : 최종 `buy=True`일 때 메시지 맨 앞에 부착
- `신호(n):` 뒤에 활성화된 신호 목록
- 볼린저/RSI/MACD/MA/거래량/Breakout/스토캐스틱/MFI/OBV/ATR 각각 표시
해당 전략으로 **과매도 바닥근처 매수 기회 + 상승 추세 전환 브레이크아웃** 두 영역을 모두 포착할 수 있습니다.
@@ -66,7 +80,7 @@ else:
## 스케줄 테이블 (기본값)
| 대상 | 실행 시각(서버 기준) | 호출 함수 |
|------|----------------------|-----------|
| KRW 코인 | 매시간 04, 34분 | `monitor_coins()` |
| KRW 코인 | 매시간 04, 14, 24, 34, 44, 54분 | `monitor_coins()` |
| 미국 주식 / ETF | 05:10, 16:30, 23:30 | `monitor_us_stocks()` |
| 한국 ETF / 주식 | 07:10, 18:20 | `monitor_kr_stocks()` |

3
config.py
View File

@@ -27,9 +27,12 @@ KR_COINS = {
"ADA": "ADA",
"APE": "ApeCoin",
"ARB": "Arbitrum",
"BONK": "BONK",
"HBAR": "HBAR",
"LINK": "Chainlink",
"ONDO": "ONDO",
"PEPE": "PEPE",
"SEI": "SEI",
"SHIB": "Shiba Inu",
"STORJ": "Storj",
"SUI": "Sui Network",

1
requirements.txt
View File

@@ -1,5 +1,6 @@
yfinance
pandas
mplcursors
numpy
ccxt
PyJWT

420
stock_monitor.py
View File

@@ -1,7 +1,6 @@
import pandas as pd
from HTS2 import HTS
import pandas as pd
from dateutil.relativedelta import relativedelta
from datetime import datetime, timedelta
import telegram
import time
@@ -12,9 +11,11 @@ from multiprocessing import Pool
import schedule
from config import *
import FinanceDataReader as fdr
import numpy as np
hts = HTS()
BUY_AMOUNT = 100000
BUY_AMOUNT = 10000
def send_coin_msg(text):
coin_client = telegram.Bot(token=COIN_TELEGRAM_BOT_TOKEN)
@@ -38,10 +39,11 @@ def send_coin_telegram_message(message_list, header):
return
def buy_ticker(buy_ticker_list):
for buy_ticker in buy_ticker_list:
ticker_code = buy_ticker['symbol']
_ = hts.buyCoinMarket(ticker_code, BUY_AMOUNT)
def buy_ticker(symbole):
try:
_ = hts.buyCoinMarket(symbole, BUY_AMOUNT)
except Exception as e:
print(f"Error buying {symbole}: {str(e)}")
return
@@ -69,232 +71,101 @@ def send_stock_telegram_message(message_list, header):
return
def calculate_bollinger_bands(data):
data['MA'] = data['Close'].rolling(window=BOLLINGER_PERIOD).mean()
data['STD'] = data['Close'].rolling(window=BOLLINGER_PERIOD).std()
data['Upper'] = data['MA'] + (BOLLINGER_STD * data['STD'])
data['Lower'] = data['MA'] - (BOLLINGER_STD * data['STD'])
return data
def normalize_data(data):
"""데이터 정규화 함수 - 모든 코인에 동일하게 적용"""
# Min-Max 정규화를 위한 컬럼
columns_to_normalize = ['Open', 'High', 'Low', 'Close', 'Volume']
normalized_data = data.copy()
# 각 컬럼별 정규화 (20일 롤링 윈도우 사용)
for column in columns_to_normalize:
min_val = data[column].rolling(window=20).min()
max_val = data[column].rolling(window=20).max()
# 0으로 나누기 방지
denominator = max_val - min_val
normalized_data[f'{column}_Norm'] = np.where(
denominator != 0,
(data[column] - min_val) / denominator,
0.5 # 기본값 설정
)
return normalized_data
def calculate_technical_indicators(data):
# 볼린저 밴드 계산
data = calculate_bollinger_bands(data)
# RSI 계산 (14일 기준)
delta = data['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
data['RSI'] = 100 - (100 / (1 + rs))
# MACD 계산
exp1 = data['Close'].ewm(span=12, adjust=False).mean()
exp2 = data['Close'].ewm(span=26, adjust=False).mean()
data['MACD'] = exp1 - exp2
data['Signal'] = data['MACD'].ewm(span=9, adjust=False).mean()
data['MACD_Hist'] = data['MACD'] - data['Signal']
# 이동평균선
"""기술적 지표 계산 - 모든 코인에 동일하게 적용"""
# 데이터 정규화
data = normalize_data(data)
# 이동평균선 계산
data['MA5'] = data['Close'].rolling(window=5).mean()
data['MA20'] = data['Close'].rolling(window=20).mean()
data['MA40'] = data['Close'].rolling(window=40).mean()
data['MA60'] = data['Close'].rolling(window=60).mean()
data['MA120'] = data['Close'].rolling(window=120).mean()
data['MA200'] = data['Close'].rolling(window=200).mean()
data['MA240'] = data['Close'].rolling(window=240).mean()
data['MA720'] = data['Close'].rolling(window=720).mean()
data['MA1440'] = data['Close'].rolling(window=1440).mean()
# 거래량 이동평균
data['Volume_MA5'] = data['Volume'].rolling(window=5).mean()
# 매수 타이밍을 이동평균선으로 결정
# 골든크로스: 단기 이동평균선이 장기 이동평균선을 상향 돌파할 때 매수
data['golden_cross'] = (data['MA5'] > data['MA20']) & (data['MA5'].shift(1) <= data['MA20'].shift(1))
# 볼린저 밴드 계산
data['MA'] = data['Close'].rolling(window=20).mean()
data['STD'] = data['Close'].rolling(window=20).std()
data['Upper'] = data['MA'] + (2 * data['STD'])
data['Lower'] = data['MA'] - (2 * data['STD'])
return data
def check_buy_signals(symbol, data):
if len(data) < 60: # 최소 60일치 데이터 필요
return None
def check_buy_point(data, simulation=None):
latest = data.iloc[-1]
prev = data.iloc[-2]
# 8월 3일 이후의 데이터만 고려
if latest.name <= pd.Timestamp('2025-08-03'):
return None
# 볼린저 밴드 신호
bb_signal = False
if isinstance(latest['Upper'], float):
upper_band = latest['Upper']
lower_band = latest['Lower']
current_price = latest['Close']
# 매수 포인트 탐지 및 표시
if simulation:
recent_data = data
else:
upper_band = latest['Upper'].iloc[0]
lower_band = latest['Lower'].iloc[0]
current_price = latest['Close'].iloc[0]
recent_data = data.tail(10)
distance = (current_price - lower_band) / (upper_band - lower_band)
bb_signal = distance < BOLLINGER_THRESHOLD
recent_data['buy_point'] = 0
for i in range(1, len(recent_data)):
if all(recent_data[f'MA{n}'].iloc[i] < recent_data['MA720'].iloc[i] for n in [5, 20, 40, 120, 200, 240]) and \
all(recent_data[f'MA{n}'].iloc[i] > recent_data[f'MA{n}'].iloc[i-1] for n in [5, 20, 40, 120, 200, 240]) and \
recent_data['MA720'].iloc[i] < recent_data['MA1440'].iloc[i]:
recent_data.at[recent_data.index[i], 'buy_point'] = 1
# U자 반등 후 이전 고점 돌파 여부 계산 (BREAKOUT)
breakout_signal = False
if len(data) >= max(BREAKOUT_LOOKBACK, BREAKOUT_WEEK_LOOKBACK) + 1:
# ① U자 형태 확인
window_close = data['Close'].iloc[-BREAKOUT_LOOKBACK - 1:-1]
prev_high = window_close.max()
prev_low = window_close.min()
if not simulation:
if recent_data['buy_point'][-10:-1].sum() > 0:
recent_data['buy_point'][-1] = 1
# ② 1주일(42캔들) 전 가격 대비 5% 이상 상승하지 않았는지 체크
price_week_ago = data['Close'].iloc[-BREAKOUT_WEEK_LOOKBACK - 1]
if price_week_ago > 0:
week_change = (current_price - price_week_ago) / price_week_ago
else:
week_change = 1 # 값이 0이면 조건 불충족 처리
return recent_data
# ③ 조건 종합: U자+돌파 && 주간 상승률 ≤ 5%
if (
prev_high > 0 and (prev_high - prev_low) / prev_high > BUY_THRESHOLD and current_price > prev_high
and week_change <= BREAKOUT_WEEK_LIMIT
):
breakout_signal = True
# 장기간 저항선 돌파 여부 계산 (LONG RESISTANCE BREAKOUT)
long_breakout_signal = False
if len(data) >= RESISTANCE_LOOKBACK + 1:
resistance_level = data['Close'].iloc[-RESISTANCE_LOOKBACK - 1:-1].max()
previous_closes = data['Close'].iloc[-RESISTANCE_LOOKBACK - 1:-1]
# 과거 구간에서 저항선 이상으로 종가가 한번도 올라가지 않은 경우 + 현재 가격이 저항선 돌파
if (previous_closes <= resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD)).all() and \
current_price > resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD):
long_breakout_signal = True
# RSI 과매도 신호 (RSI < 30)
if not isinstance(latest['Upper'], float):
rsi_signal = latest['RSI'].iloc[0] < 30
# MACD 신호 (MACD가 시그널 라인을 상향 돌파)
macd_signal = (prev['MACD'].iloc[0] < prev['Signal'].iloc[0]) and (
latest['MACD'].iloc[0] > latest['Signal'].iloc[0])
# 이동평균선 골든크로스 임박 또는 발생
ma_signal = (prev['MA5'].iloc[0] < prev['MA20'].iloc[0]) and (latest['MA5'].iloc[0] >= latest['MA20'].iloc[0])
# 거래량 증가 신호 (5일 평균 대비 150% 이상)
volume_signal = latest['Volume'].iloc[0] > (latest['Volume_MA5'].iloc[0] * 1.5)
# 종합 신호
buy_signals = {
'bb_signal': bb_signal,
'rsi_signal': rsi_signal,
'macd_signal': macd_signal,
'ma_signal': ma_signal,
'volume_signal': volume_signal,
'breakout_signal': breakout_signal,
'long_breakout_signal': long_breakout_signal
}
# 최소 3개 이상의 신호가 동시에 발생할 때 매수 신호로 간주
signal_count = sum(1 for signal in buy_signals.values() if signal)
return {
'symbol': symbol,
'price': current_price,
'lower_band': lower_band,
'distance': distance,
'rsi': latest['RSI'].iloc[0],
'macd': latest['MACD'].iloc[0],
'signal_line': latest['Signal'].iloc[0],
'buy_signals': buy_signals,
'signal_count': signal_count,
'buy': long_breakout_signal or breakout_signal or (
(bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal)))
}
else:
rsi_signal = latest['RSI'] < 30
# MACD 신호 (MACD가 시그널 라인을 상향 돌파)
macd_signal = (prev['MACD'] < prev['Signal']) and (latest['MACD'] > latest['Signal'])
# 이동평균선 골든크로스 임박 또는 발생
ma_signal = (prev['MA5'] < prev['MA20']) and (latest['MA5'] >= latest['MA20'])
# 거래량 증가 신호 (5일 평균 대비 150% 이상)
volume_signal = latest['Volume'] > (latest['Volume_MA5'] * 1.5)
# 종합 신호
buy_signals = {
'bb_signal': bb_signal,
'rsi_signal': rsi_signal,
'macd_signal': macd_signal,
'ma_signal': ma_signal,
'volume_signal': volume_signal,
'breakout_signal': breakout_signal,
'long_breakout_signal': long_breakout_signal
}
# 최소 3개 이상의 신호가 동시에 발생할 때 매수 신호로 간주
signal_count = sum(1 for signal in buy_signals.values() if signal)
return {
'symbol': symbol,
'price': current_price,
'lower_band': lower_band,
'distance': distance,
'rsi': latest['RSI'],
'macd': latest['MACD'],
'signal_line': latest['Signal'],
'buy_signals': buy_signals,
'signal_count': signal_count,
'buy': long_breakout_signal or breakout_signal or (
(bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal)))
}
def format_message(info, market_type):
message = ""
if info['buy']:
message += '🛒 '
message += f"[{market_type}] {info['name']} ({info['symbol']}) "
message += f"현재가: {'$' if market_type == 'US' else '₩'}{info['price']:.2f}, "
# 매수 신호 상세 정보
count = 0
if any(info['buy_signals'].values()):
message += "📊신호 ({count}):"
if info['buy_signals']['bb_signal']:
message += "- 볼린저 밴드 하단 근접 (근접도: {:.1f}%),".format(info['distance'] * 100)
count += 1
if info['buy_signals']['rsi_signal']:
message += f"- RSI 과매도 구간 (RSI: {info['rsi']:.1f}),"
count += 1
if info['buy_signals']['macd_signal']:
message += "- MACD 골든크로스,"
count += 1
if info['buy_signals']['ma_signal']:
message += "- 이동평균선 골든크로스,"
count += 1
if info['buy_signals']['volume_signal']:
message += "- 거래량 급증"
count += 1
if info['buy_signals'].get('breakout_signal'):
message += "- U자 반등 돌파"
count += 1
if info['buy_signals'].get('long_breakout_signal'):
message += "- 장기 저항 돌파"
count += 1
message += "\n"
message = message.replace("{count}", str(count))
def format_message(market_type, symbol, symbol_name, close):
message = f"매수 [{market_type}] {symbol_name} ({symbol}) "
message += f"현재가: {'$' if market_type == 'US' else '₩'}{close:.2f}, "
return message
def format_ma_message(info, market_type):
"""MA 알림 메시지 생성"""
prefix = '📈 ' if info.get('alert') else ''
prefix = '상승 ' if info.get('alert') else ''
message = prefix + f"[{market_type}] {info['name']} ({info['symbol']}) "
message += f"현재가: {'$' if market_type == 'US' else '₩'}{info['price']:.2f} \n"
return message
def get_coin_data(symbol, interval=240, retries=3):
def get_coin_data(symbol, interval=240, to=None, retries=3):
for attempt in range(retries):
try:
url = "https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=3000".format(interval, symbol)
#url = "https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=3000".format(interval, symbol)
if to is None:
url = ("https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=200").format(interval, symbol)
else:
url = ("https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=200&to={}").format(interval, symbol, to)
#url = 'https://api.bithumb.com/v1/candles/minutes/60?market=KRW-ADA&count=200'
#url = 'https://api.bithumb.com/v1/candles/minutes/minutes/60?market=KRW-ADA&count=200&to=2025-08-06 10:38:38'
headers = {"accept": "application/json"}
response = requests.get(url, headers=headers)
json_data = json.loads(response.text)
@@ -328,6 +199,27 @@ def get_coin_data(symbol, interval=240, retries=3):
continue
return None
def get_coin_more_data(symbol, interval, bong_count=3000):
# 코인 데이터 1500개 봉 가져오기
to = datetime.now()
data = None
while data is None or len(data) < bong_count:
if data is None:
data = get_coin_data(symbol, interval, to.strftime("%Y-%m-%d %H:%M:%S"))
else:
df = get_coin_data(symbol, interval, to.strftime("%Y-%m-%d %H:%M:%S"))
data = pd.concat([data, df], ignore_index=True)
time.sleep(0.3)
to = to - relativedelta(minutes=interval * 200)
data = data.set_index('datetime')
data = data.sort_index()
data = data.drop_duplicates(keep='first')
data["datetime"] = data.index
# 코인 데이터 1500개 봉 가져오기
return data
def get_kr_stock_data(symbol, retries=3):
for attempt in range(retries):
@@ -368,14 +260,11 @@ def monitor_us_stocks():
if data is not None and not data.empty:
try:
data = calculate_technical_indicators(data)
info = detect_turnaround_signal(symbol, data, 0)
if info is None:
recent_data = check_buy_point(data) # Changed to check_buy_point
if recent_data['buy_point'][-1] != 1:
continue
info['name'] = US_STOCKS[symbol]
print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']}")
if info['alert']:
message_list.append(format_ma_message(info, 'US'))
print(f" - {US_STOCKS[symbol]} ({symbol}): {recent_data['Close'][-1]:.2f}")
message_list.append(format_message('US', symbol, US_STOCKS[symbol], recent_data['Close'][-1]))
except Exception as e:
print(f"Error processing data for {symbol}: {str(e)}")
time.sleep(0.5)
@@ -402,14 +291,11 @@ def monitor_kr_stocks():
if data is not None and not data.empty:
try:
data = calculate_technical_indicators(data)
info = detect_turnaround_signal(symbol, data, 0)
if info is None:
recent_data = check_buy_point(data) # Changed to check_buy_point
if recent_data['buy_point'][-1] != 1:
continue
info['name'] = KR_ETFS[symbol]
print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']}")
if info['alert']:
message_list.append(format_ma_message(info, 'KR'))
print(f" - {KR_ETFS[symbol]} ({symbol}): {recent_data['Close'][-1]:.2f}")
message_list.append(format_message('KR', symbol, US_STOCKS[symbol], recent_data['Close'][-1]))
except Exception as e:
print(f"Error processing data for {symbol}: {str(e)}")
@@ -434,25 +320,25 @@ def monitor_kr_stocks():
def monitor_coins():
message_list = []
buy_ticker_list = []
print("KRW COINs {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
for symbol in KR_COINS:
# 1시간
interval = 60
data = get_coin_data(symbol, interval=interval)
data = get_coin_more_data(symbol, interval)
if data is not None and not data.empty:
try:
data = calculate_technical_indicators(data)
info = detect_turnaround_signal(symbol, data, interval)
if info is None:
recent_data = check_buy_point(data) # Changed to check_buy_point
if recent_data['buy_point'][-1] != 1:
continue
info['name'] = KR_COINS[symbol]
print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})")
print(f" - {KR_ETFS[symbol]} ({symbol}): {recent_data['Close'][-1]:.2f}")
message_list.append(format_message('COIN', symbol, US_STOCKS[symbol], recent_data['Close'][-1]))
# buy
buy_ticker(symbol)
if info['alert']:
message_list.append(format_ma_message(info, 'KR'))
buy_ticker_list.append(info)
except Exception as e:
print(f"Error processing data for {symbol}: {str(e)}")
else:
@@ -461,18 +347,19 @@ def monitor_coins():
# 4시간
interval = 240
data = get_coin_data(symbol, interval=interval)
data = get_coin_more_data(symbol, interval, bong_count=1500)
if data is not None and not data.empty:
try:
data = calculate_technical_indicators(data)
info = detect_turnaround_signal(symbol, data, interval)
info = check_buy_point(data, simulation=True) # Changed to check_buy_point
if info is None:
continue
info['name'] = KR_COINS[symbol]
print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})")
print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['buy']}")
if info['alert']:
message_list.append(format_ma_message(info, 'KR'))
if info['buy']:
message_list.append(format_message(info, 'KR'))
except Exception as e:
print(f"Error processing data for {symbol}: {str(e)}")
else:
@@ -481,8 +368,6 @@ def monitor_coins():
if len(message_list) > 0:
try:
# buy
buy_ticker(buy_ticker_list)
# send message
send_coin_telegram_message(message_list, header="[KRW-COIN]")
except Exception as e:
@@ -494,70 +379,19 @@ def monitor_coins():
# Turnaround Detector v6
# ----------------------
def detect_turnaround_signal(symbol, data, interval=0):
def detect_turnaround_signal(symbol, data, interval=0, params=None):
if len(data) < 7:
return None
cur = data.iloc[-1]
# 이동평균을 기반으로 매수 신호 결정
cur = data.iloc[-1]
prev = data.iloc[-2]
recent = data.iloc[-7:-1]
# ①~② 국지 바닥 + 하단 터치
is_bottom = prev.Low == recent['Low'].min()
touch_lower = prev.Close <= prev.Lower * 1.01
# ③ 첫 반등 + MA5 돌파
rebound = (cur.Close >= prev.Close * 1.005) and (cur.Close >= cur.Lower * 1.01)
ma5_break = (prev.Close <= prev.MA5) and (cur.Close > cur.MA5)
ma20_below = cur.Close < cur.MA20
# ④ RSI 회복
rsi_recover = (prev.RSI < 40) and (cur.RSI > 40) and (cur.RSI > prev.RSI)
# ⑤ MACD 히스토그램 양전환
hist_cross = (prev.MACD_Hist <= 0) and (cur.MACD_Hist > 0)
# ⑥ MA5-MA20 골든크로스 (이번 봉) (or DMI 사용)
cross_ma = (prev.MA5 <= prev.MA20) and (cur.MA5 > cur.MA20)
# ⑦ 장기 추세 아직 하락
down_trend = cur.MA20 < cur.MA40
# 최근 저점 탐지 강화
recent_low = data['Low'].iloc[-10:].min()
is_recent_low = cur.Low <= recent_low * 1.005
# 추가 조건: MA5가 MA20을 상향 돌파
ma5_cross_ma20 = (prev.MA5 <= prev.MA20) and (cur.MA5 > cur.MA20)
# 매수 신호 조건 (저점 + 단기 MA 돌파 + RSI 회복)
# 볼린저 밴드 포지션
distance = (cur.Close - cur.Lower) / (cur.Upper - cur.Lower)
# 매수 신호 조건: 전봉 하단 밴드 터치 + 반등 + MA5 돌파 + RSI 회복 + 밴드 위치 ≤ 0.35
alert = touch_lower and rebound and ma5_break and rsi_recover and (distance <= 0.35)
return {
"symbol": symbol,
"price": float(cur.Close),
"alert": alert,
"details": {
"interval": interval,
"is_bottom": is_bottom,
"touch": touch_lower,
"rebound": rebound,
"ma5_break": ma5_break,
"ma20_below": ma20_below,
"rsi_recover": rsi_recover,
"hist_cross": hist_cross,
"cross_ma": cross_ma,
"down_trend": down_trend,
},
}
return None
def run_schedule():
monitor_coins()
# 코인 모니터링 스케줄 (매시간 1분, 11분, 21분, 31분, 41분, 51분)
for minute in [4, 14, 24, 34, 44, 54]:
schedule.every().hour.at(f":{minute:02d}").do(monitor_coins)

261
stock_simulation.py
View File

@@ -1,14 +1,16 @@
import math
import requests
import time
from datetime import datetime
from dateutil.relativedelta import relativedelta
import pandas as pd
import yfinance as yf
import matplotlib.pyplot as plt
import matplotlib.dates
import mplcursors
plt.rcParams['font.family'] ='AppleGothic'
plt.rcParams['axes.unicode_minus'] =False
from config import *
from stock_monitor import calculate_technical_indicators, detect_turnaround_signal
from stock_monitor import calculate_technical_indicators, detect_turnaround_signal, get_coin_more_data, check_buy_point
# 비트/알트코인 KRW 마켓 식별: 문자열 "-KRW" 포함 여부로 간단 구분
@@ -19,50 +21,12 @@ INTERVAL_MAP = {
BITHUMB_MAX_COUNT = 3000 # API 최대 3000 캔들
def fetch_price_history(symbol: str, interval_minutes: int, days: int = 30) -> pd.DataFrame:
def fetch_coin_history_bithumb(symbol: str, interval_minutes: int, days: int) -> pd.DataFrame:
"""빗썸 API를 이용해 최근 `days`일 코인 데이터 수집 (interval 60 / 240)"""
if interval_minutes not in (60, 240):
raise ValueError("Bithumb API only supports 60 or 240 minutes in this helper")
minutes = interval_minutes
count = int(math.ceil(days * 24 * 60 / minutes)) + 10 # 여유분 10 캔들
count = min(count, BITHUMB_MAX_COUNT)
url = f"https://api.bithumb.com/v1/candles/minutes/{minutes}?market=KRW-{symbol}&count={count}"
res = requests.get(url, timeout=5)
res.raise_for_status()
raw = res.json()
if not isinstance(raw, list) or len(raw) == 0:
raise RuntimeError("Empty response from Bithumb API")
df_temp = pd.DataFrame(raw)
# API 반환: [timestamp, open, close, high, low, volume] 순
df_temp = df_temp.sort_index(ascending=False) # 최신순, 뒤집어서 역순 전달
data = pd.DataFrame()
# data.columns = ['datetime', 'open', 'close', 'high', 'low', 'volume']
# data['datetime'] = pd.to_datetime(data_temp['candle_date_time_kst'])
data['datetime'] = pd.to_datetime(df_temp['candle_date_time_kst'], format='%Y-%m-%dT%H:%M:%S')
data['Open'] = df_temp['opening_price']
data['Close'] = df_temp['trade_price']
data['High'] = df_temp['high_price']
data['Low'] = df_temp['low_price']
data['Volume'] = df_temp['candle_acc_trade_volume']
data = data.set_index('datetime')
data = data.astype(float)
data["datetime"] = data.index
data = data.set_index("datetime").sort_index() # 시간 오름차순
return data
def fetch_price_history(symbol: str, interval_minutes: int, days: int = 7) -> pd.DataFrame:
"""최근 `days`일 데이터(캔들)를 가져온다. 코인(-KRW)은 빗썸, 그 외 yfinance."""
if symbol in KR_COINS:
base_symbol = symbol.replace("-KRW", "")
return fetch_coin_history_bithumb(base_symbol, interval_minutes, days)
bong_count = 3000
return get_coin_more_data(symbol, interval_minutes, bong_count=bong_count)
# -------- 주식/ETF/해외코인 (yfinance) --------
if interval_minutes not in INTERVAL_MAP:
@@ -83,47 +47,198 @@ def fetch_price_history(symbol: str, interval_minutes: int, days: int = 7) -> pd
return df
def run_simulation(symbol: str, interval_minutes: int, days: int = 7):
def analyze_bottom_period(symbol: str, interval_minutes: int, days: int = 90):
"""저점 기간(6월 22일~7월 9일) 분석"""
data = fetch_price_history(symbol, interval_minutes, days)
data = calculate_technical_indicators(data)
alerts = [] # (timestamp, price)
# 시계열 순회하며 알림 조건 체크
for i in range(len(data)):
slice_df = data.iloc[: i + 1]
print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}")
print(f"총 데이터 수: {len(data)}")
# 저점 기간 필터링 (6월 22일~7월 9일)
bottom_start = pd.Timestamp('2025-06-22')
bottom_end = pd.Timestamp('2025-07-09')
bottom_data = data[(data.index >= bottom_start) & (data.index <= bottom_end)]
if len(bottom_data) == 0:
print("저점 기간 데이터가 없습니다.")
return
print(f"\n저점 기간 데이터: {bottom_data.index[0]} ~ {bottom_data.index[-1]}")
print(f"저점 기간 데이터 수: {len(bottom_data)}")
# 저점 기간의 기술적 지표 분석
print("\n=== 저점 기간 기술적 지표 분석 ===")
# 1. 가격 분석
min_price = bottom_data['Low'].min()
max_price = bottom_data['High'].max()
avg_price = bottom_data['Close'].mean()
print(f"최저가: {min_price:.4f}")
print(f"최고가: {max_price:.4f}")
print(f"평균가: {avg_price:.4f}")
print(f"가격 변동폭: {((max_price - min_price) / min_price * 100):.2f}%")
# 3. 볼린저 밴드 분석
bb_lower_min = bottom_data['Lower'].min()
bb_upper_max = bottom_data['Upper'].max()
print(f"\n볼린저 밴드 분석:")
print(f"하단 밴드 최저: {bb_lower_min:.4f}")
print(f"상단 밴드 최고: {bb_upper_max:.4f}")
# 4. 거래량 분석
volume_avg = bottom_data['Volume'].mean()
volume_max = bottom_data['Volume'].max()
print(f"\n거래량 분석:")
print(f"평균 거래량: {volume_avg:.0f}")
print(f"최대 거래량: {volume_max:.0f}")
# 5. 실제 저점 찾기
actual_bottom_idx = bottom_data['Low'].idxmin()
actual_bottom_price = bottom_data.loc[actual_bottom_idx, 'Low']
actual_bottom_date = actual_bottom_idx
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. 매수 신호 분석
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]
info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes)
if info and info["alert"]:
alerts.append((slice_df.index[-1], slice_df["Close"].iloc[-1]))
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]))
print(f"저점 기간 매수 신호 수: {len(alerts)}")
if alerts:
print("매수 신호 발생 시점:")
for date, price in alerts:
print(f" {date}: {price:.4f}")
return bottom_data, alerts
# 모든 매수 신호를 표시
# 기존 필터 제거하여 전체 기간 매수 신호 사용
def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
data = fetch_price_history(symbol, interval_minutes)
data = calculate_technical_indicators(data)
data = check_buy_point(data, simulation=True)
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}, # 매우 엄격한 설정
]
alerts = []
for params in param_candidates:
alerts.clear()
for i in range(len(data)):
slice_df = data.iloc[: i + 1]
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]))
print(f"\n총 매수 신호 수: {len(alerts)}")
# 서브플롯 생성
fig, ax1 = plt.subplots(figsize=(15, 8))
fig.suptitle(f"{symbol} - 시뮬레이션 {interval_minutes}분봉", fontsize=14)
# 메인 차트 (가격, 이동평균선, 볼린저 밴드)
line_close = ax1.plot(data.index, data["Close"], label="종가", color="black", linewidth=1.5)[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]
line_ma120 = ax1.plot(data.index, data["MA120"], label="MA120", color="purple", linewidth=1)[0]
line_ma200 = ax1.plot(data.index, data["MA200"], label="MA200", color="brown", linewidth=1)[0]
line_ma240 = ax1.plot(data.index, data["MA240"], label="MA240", color="black", linewidth=1)[0]
line_ma720 = ax1.plot(data.index, data["MA720"], label="MA720", color="cyan", linewidth=1)[0]
line_ma1440 = ax1.plot(data.index, data["MA1440"], label="MA1440", color="magenta", linewidth=1)[0]
# Plot
plt.figure(figsize=(12, 6))
plt.plot(data.index, data["Close"], label="종가", color="black")
plt.plot(data.index, data["MA5"], label="MA5", color="orange", linewidth=1)
plt.plot(data.index, data["MA20"], label="MA20", color="blue", linewidth=1)
plt.plot(data.index, data["MA40"], label="MA40", color="green", linewidth=1)
# Bollinger Bands
plt.plot(data.index, data["Upper"], label="볼린저 Upper", color="grey", linestyle="--", linewidth=1)
plt.plot(data.index, data["Lower"], label="볼린저 Lower", color="grey", linestyle="--", linewidth=1)
plt.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1)
line_upper = ax1.plot(data.index, data["Upper"], label="볼린저 Upper", color="grey", linestyle="--", linewidth=1)[0]
line_lower = ax1.plot(data.index, data["Lower"], label="볼린저 Lower", color="grey", linestyle="--", linewidth=1)[0]
ax1.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1)
# 매수 신호 표시
scatter_buy = None
if alerts:
times, prices = zip(*alerts)
plt.scatter(times, prices, facecolors='none', edgecolors='red', linewidths=2, s=150, zorder=6, label='매수신호')
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
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_point'] = 1
# 매수 포인트를 빨간 동그라미로 표시
buy_points = data[data['buy_point'] == 1]
scatter_buy_points = ax1.scatter(buy_points.index, buy_points['Close'], color='red', s=100, zorder=5, label='매수 포인트')
# 마우스 오버 기능 추가
cursor = mplcursors.cursor(scatter_buy_points, hover=True)
cursor.connect("add", lambda sel: sel.annotation.set_text(
f'날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}'
))
cursor.connect("remove", lambda sel: sel.annotation.set_visible(False))
ax1.set_ylabel("가격")
ax1.legend(loc='upper left', fontsize=10)
ax1.grid(True, linestyle='--', alpha=0.5)
plt.title(f"{symbol} – 시뮬레이션 {interval_minutes}분봉 (최근 {days}일)")
plt.xlabel("날짜")
plt.ylabel("가격")
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
if __name__ == "__main__":
symbol = 'WLD'
interval = 60
days = 7
run_simulation(symbol, interval, days)
days = 90 # 분석 기간을 90일로 늘림 (6월~8월 데이터 포함)
target_coins = ['ADA','APE','ARB','BONK','HBAR','LINK','ONDO','PEPE','SEI','SHIB','STORJ','SUI','TON','TRX','WLD','XLM','XRP']
#target_coins = ['APE']
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)
except Exception as e:
print(f"Error analyzing {symbol}: {str(e)}")