111 lines
4.7 KiB
Python
111 lines
4.7 KiB
Python
import pandas as pd
|
|
from datetime import datetime, timedelta
|
|
import time
|
|
import schedule
|
|
from config import *
|
|
import FinanceDataReader as fdr
|
|
|
|
from monitor_min import Monitor
|
|
|
|
class MonitorStock (Monitor):
|
|
"""자산(코인/주식/ETF) 모니터링 및 매수 실행 클래스"""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(None)
|
|
|
|
|
|
def get_kr_stock_data(self, symbol: str, retries: int = 3) -> pd.DataFrame | None:
|
|
for attempt in range(retries):
|
|
try:
|
|
end = datetime.now()
|
|
start = end - timedelta(days=300)
|
|
data = fdr.DataReader(symbol, start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d'))
|
|
if not data.empty:
|
|
data = data.rename(columns={
|
|
'Open': 'Open',
|
|
'High': 'High',
|
|
'Low': 'Low',
|
|
'Close': 'Close',
|
|
'Volume': 'Volume',
|
|
})
|
|
return data
|
|
print(f"No data received for {symbol}, attempt {attempt + 1}")
|
|
time.sleep(2)
|
|
except Exception as e:
|
|
print(f"Attempt {attempt + 1} failed for {symbol}: {str(e)}")
|
|
if attempt < retries - 1:
|
|
time.sleep(5)
|
|
continue
|
|
return None
|
|
|
|
# ------------- Monitors -------------
|
|
def monitor_us_stocks(self) -> None:
|
|
message_list: list[str] = []
|
|
print("US Stocks {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
|
|
for symbol in US_STOCKS:
|
|
data = self.get_kr_stock_data(symbol)
|
|
if data is not None and not data.empty:
|
|
try:
|
|
data = self.calculate_technical_indicators(data)
|
|
recent_data = self.check_point(symbol, data)
|
|
if recent_data['point'].iloc[-1] != 1:
|
|
continue
|
|
print(f" - {US_STOCKS[symbol]} ({symbol}): {recent_data['Close'].iloc[-1]:.2f}")
|
|
message_list.append(
|
|
self.format_message('US', symbol, US_STOCKS[symbol], recent_data['Close'].iloc[-1], recent_data['signal'].iloc[-1])
|
|
)
|
|
except Exception as e:
|
|
print(f"Error processing data for {symbol}: {str(e)}")
|
|
time.sleep(0.5)
|
|
if len(message_list) > 0:
|
|
try:
|
|
self.send_stock_telegram_message(message_list, header="[US-STOCK]")
|
|
except Exception as e:
|
|
print(f"Error sending Telegram message: {str(e)}")
|
|
|
|
def monitor_kr_stocks(self) -> None:
|
|
message_list: list[str] = []
|
|
print("KR ETFs {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
|
|
for symbol in KR_ETFS:
|
|
try:
|
|
clean_symbol = symbol.replace('.KS', '')
|
|
data = self.get_kr_stock_data(clean_symbol)
|
|
if data is not None and not data.empty:
|
|
try:
|
|
data = self.calculate_technical_indicators(data)
|
|
recent_data = self.check_point(symbol, data)
|
|
if recent_data['point'].iloc[-1] != 1:
|
|
continue
|
|
print(f" - {KR_ETFS[symbol]} ({symbol}): {recent_data['Close'].iloc[-1]:.2f}")
|
|
message_list.append(
|
|
self.format_message('KR', symbol, KR_ETFS[symbol], recent_data['Close'].iloc[-1], recent_data['signal'].iloc[-1])
|
|
)
|
|
except Exception as e:
|
|
print(f"Error processing data for {symbol}: {str(e)}")
|
|
else:
|
|
print(f"Data for {symbol} is empty or None.")
|
|
time.sleep(1)
|
|
except Exception as e:
|
|
print(f"Unexpected error processing {symbol}: {str(e)}")
|
|
continue
|
|
if len(message_list) > 0:
|
|
try:
|
|
self.send_stock_telegram_message(message_list, header="[KR-STOCK]")
|
|
except Exception as e:
|
|
print(f"Error sending Telegram message: {str(e)}")
|
|
|
|
# ------------- Scheduler -------------
|
|
def run_schedule(self) -> None:
|
|
schedule.every().day.at("16:30").do(self.monitor_us_stocks)
|
|
schedule.every().day.at("23:30").do(self.monitor_us_stocks)
|
|
schedule.every().day.at("08:10").do(self.monitor_kr_stocks)
|
|
schedule.every().day.at("18:20").do(self.monitor_kr_stocks)
|
|
print("Scheduler started. Stock Monitoring will run at specified times.")
|
|
while True:
|
|
schedule.run_pending()
|
|
time.sleep(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
MonitorStock().run_schedule()
|