[Python] 봇 개선 및 커맨드 추가

새로운 기능으로 주식 정보를 가져오는 명령어를 만드려고 했다가 아주 큰일이 됐다.

굳이 손대고 싶지 않았던 귀찮은 부분들에 손을 대기로 했다.

 

 

1. 메인 봇 코드 다이어트

기존엔 명령어를 별도로 빼는 것이 아니라, 아래처럼 그 명령어 함수 안에 구동부를 다 때려 넣은 형식이었다.

 

#region sync
@client.tree.command(name='sync', description='Owner only')
async def sync(interaction: discord.Interaction):
    if interaction.user.id == OWNER_ID:
        await interaction.response.defer(ephemeral=True)
        await client.tree.sync()
        await interaction.followup.send('Synced', ephemeral=True)
    else:
        await interaction.followup.send('You have no permission to use this command.', ephemeral=True)
#endregion

 

일부 명령어의 경우엔 이 길이가 매우 길어서 매우 보기 좋지 않았고 유지보수성도 떨어졌다.

이걸 이제 executeSync()와 같은 형식으로 바꾸고 별도 파일로 다 빼놨다는 소리.

 

명령어는 이렇게 정리해 놨다.

짧아지니 보기는 좋다.

 

2. Epoch Converting 개선

개선이라 해놨지만 사실상 이전엔 완전히 오답이었고 이제 정답을 찾아왔다는 느낌.

 

딱 이거다

이전엔 Naive로 어거지로 끼워 맞췄다면, 이젠 제대로 Aware로 동작한다.

클라우드는 UTC로 돌아가기 때문에 여기와는 시간이 다를 수밖에.

그걸 억지로 보정하려 한 것이 기존 방식이었다.

 

from datetime import datetime
from pytz import timezone
from math import trunc

def ConvertTime(year, month, day, hour, min, sec):
    try:
        # 입력된 날짜와 시간을 datetime 객체로 생성하고 타임존을 직접 설정
        input_time = timezone('Asia/Tokyo').localize(datetime(year, month, day, hour, min, sec))
        # JST 시간을 타임스탬프로 변환
        timestamp = trunc(int(input_time.timestamp()))
        
        # ...

이젠 pytz을 활용해 제대로 된 방식으로 결과를 뱉어준다.

억지로 tz_info에 pytz을 쑤셔 넣으려고 했다가 잘못된 결과를 봤을 때 싸함을 느꼈었다...

그게 잘못된 방식이더라. 공식에서도 localize()를 통해 수행하는 것을 소개하는 걸 보면 이게 맞는 것이지 싶다.

 

 

3. /stock 커맨드 추가

뭔가 추가해 볼까 고민하다가 주식 정보를 쏴주는 기능을 추가하기로 결정했다.

주변에 투자를 하는 사람이 많기도 하고, 나 또한 투자를 진행하고 있으니 간략한 정보를 제공하는 데 괜찮지 싶었다.

 

결과물부터 보자면 아래와 같다.

 

기본적인 주가 데이터와 일부 관심도가 높은 지표들을 표시함과 동시에 차트도 제공한다.

개장 시간엔 실시간 데이터만 가져오고, 폐장 시엔 프리장 가격도 가져온다.

 

 

3-1. 준비물

1. yfinance
   Yahoo! Finance에서 정보를 가져오기 위함
2. mplfinance
   캔들 차트를 만들기 위함
3. matplotlib
   만들어진 캔들 차트를 이미지로 만들기 위함
   사실 얘가 매우 중요하다
4. pandas_market_calendars
   Nasdaq 개폐장 정보를 가져오기 위함
5. pytz
   위의 개폐장 정보를 얻기 위한 도구로 사용
6. selenium
   프리장 및 ETF 관련 정보를 스크랩하기 위함
   이 녀석 때문에 클라우드도 바꿨다
7. Red-Black Tree
   Nasdaq 상장 주식인지 아닌지 확인할 때 쓴다.

 

3-2. 구현

정보를 가져오는 것 자체는 어려울 것이 없다.

 

async def executeStock(interaction, ticker, driver):
       
    await interaction.response.defer(ephemeral=False)

    # Set default value
    if ticker is None:
        ticker = 'NVDA'
    else:
        ticker = ticker.upper()

    if check_file_exist():
        rbt.load_from_json(jsonPath)
    else:
        create_rbt_from_csv_and_save(rbt, csvPath, jsonPath)
    
    # Get stock information from Yahoo Finance API
    stock_info = yf.Ticker(ticker).info
    
    # ...

기본으로 엔비디아 정보를 가져오게 했다.

yf.Ticker(ticker).info를 통해 API에서 제공하는 주식 데이터를 받아볼 수 있다.

상당히 많은 정보가 제공되는데, 여기서 필요한 것만 뽑아서 출력하는 것이 목표.

 

# Check if the stock is an Equity or ETF
if stock_info['quoteType'] == 'EQUITY':
    stock_type = 'Equity'
elif stock_info['quoteType'] == 'ETF':
    stock_type = 'ETF'
else:
    stock_type = 'Unknown'
# Handle the stock based on its type
if stock_type == 'Equity':
    embed = handle_equity(stock_info, data, isOpen)
elif stock_type == 'ETF':
    embed = handle_etf(stock_info, data, isOpen)
else:
    embed = discord.Embed(title=f"{stock_info['longName']} / [{stock_info['symbol']}]", 
                          url=f"https://finance.yahoo.com/quote/{stock_info['symbol']}",
                          colour=discord.Colour.dark_blue(), 
                          timestamp=datetime.datetime.utcnow())
    embed.add_field(name="Error", value="Unknown stock type", inline=False)
    
#  ...

여기서 일반 주식과 ETF를 구분해 다른 프로세스를 거치게 한다.

API가 던져주는 데이터가 좀 다르기 때문에 구분할 필요가 있었다.

저렇게 분기문을 하나 더 거칠 필요는 없지만 만들다 보니 일단 저렇게 됐다.

 

# Get historical data for the past week
hist_data = yf.download(stock_info['symbol'], period="60d", interval="1d")

# Create a candlestick chart with volume bars and a moving average line
mpf.plot(hist_data,
         type='candle',
         mav=(2, 4, 6),
         volume=True,
         style=binance_dark,
         ylabel=f'Price ({stock_info["currency"]})',
         ylabel_lower='Volume',
         tight_layout=True)

# Save the plot to a buffer
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)

# Add the plot to the embed message
file = discord.File(buf, filename='plot.png')
embed.set_image(url='attachment://plot.png')

# Close the figure to prevent it from being displayed
plt.close()

await interaction.followup.send(embed=embed, file=file)

캔들 차트를 만들기 위해 데이터를 가져온다. 기간은 60일에 1일 간격으로 데이터를 가져오게 했다.

스타일을 지정하고 이미지 파일로 만들어서 메세지에 첨부해 보내면 끝.

 

이런 과정이 selenium을 사용하기 전까지의 과정이다.

저걸 사용하고 나서 좀 바뀐다.

 

3-3. Selenium

왜 이걸 쓰냐면, API가 프리장 데이터를 안 주기 때문이다.

웹에서 데이터를 JS를 활용해 실시간 갱신하기 때문에 bs4로 해당 데이터를 잡을 수가 없다.

다른 JS를 잡아준다는 라이브러리도 써봤는데 안 해주더라.

하지만 selenium은 잡아줬기 때문에 사용하기로 했다.

 

# Path to the WebDriver executable
driver_path = './Driver/chromedriver.exe'  # Update this with the path to your ChromeDriver executable

# Set up Chrome options for headless browsing
chrome_options = Options()
chrome_options.add_argument("--headless")  # Run Chrome in headless mode (no GUI)

# Initialize the Chrome driver with headless option
driver = webdriver.Chrome(executable_path=driver_path, options=chrome_options)

 

 

셀레니움 기본 세팅에 관한 건 다른 곳에 많이 있기 때문에 굳이 설명할 필요 없을 것 같다.

속도와 효율을 위해 저기에 argument를 더 추가할 수 있을 것이다.

 

# Initialize a dictionary to store the scraped data
data = {}    
driver.get(f"https://finance.yahoo.com/quote/{ticker}")

# Define the XPaths for the elements both in regular and post market
xpaths = {
    'regularMarketPrice': f'//fin-streamer[@data-field="regularMarketPrice" and @data-symbol="{ticker.upper()}"]' if stock_info['quoteType'] == 'ETF' else None,
    'regularMarketChange': f'//fin-streamer[@data-field="regularMarketChange" and @data-symbol="{ticker.upper()}"]/span' if stock_info['quoteType'] == 'ETF' else None,
    'regularMarketChangePercent': f'//fin-streamer[@data-field="regularMarketChangePercent" and @data-symbol="{ticker.upper()}"]/span' if stock_info['quoteType'] == 'ETF' else None,
    'postMarketPrice': '//fin-streamer[@data-field="postMarketPrice"]' if not isOpen else None,
    'postMarketChange': '//fin-streamer[@data-field="postMarketChange"]/span' if not isOpen else None,
    'postMarketChangePercent': '//fin-streamer[@data-field="postMarketChangePercent"]/span' if not isOpen else None
}

# Remove None values from the dictionary
xpaths = {k: v for k, v in xpaths.items() if v is not None}    

if stock_info['quoteType'] == 'ETF' or is_nasdaq_stocks(ticker):
# Loop over the XPaths and scrape the data
    for field, xpath in xpaths.items():
        try:
            element = WebDriverWait(driver, 10).until(
                EC.visibility_of_element_located((By.XPATH, xpath))
            )
            data[field] = element.text.strip()
        except Exception as e:
            Logger.error(f"Element {field} not found:", e)

위와 같이 XPath를 사용해 필요한 데이터를 뽑아온다.

이게 시간이 그냥 엄청 걸린다. VM도 무료 플랜이라 가뜩이나 성능이 안 좋아서 머리가 더 아프다.

 

여하튼 이런 식으로 뽑아온 데이터를 아래처럼 활용하는 것이다.

def handle_equity(stock_info, data, isOpen):

    isnaq = is_nasdaq_stocks(stock_info['symbol'])

    # Create embed message with stock information
    embed = discord.Embed(title=f"{stock_info['longName']} / [{stock_info['symbol']}]",
                          url=f"https://finance.yahoo.com/quote/{stock_info['symbol']}",
                          timestamp=datetime.datetime.utcnow())
    
    # get percentage and price change from openprice and current price
    price_change = stock_info['currentPrice'] - stock_info['regularMarketPreviousClose']
    percent_change = (price_change / stock_info['regularMarketPreviousClose']) * 100
            
    if price_change > 0:
        embed.colour = discord.Colour.green()
        updown = '<:yangbonghoro:>'
    elif price_change < 0:
        embed.colour = discord.Colour.red()
        updown = '<:sale:>'
    else:
        embed.colour = discord.Colour.dark_blue()
        updown = ''

    # Current Price
    embed.add_field(name="**Current Price**", value=f"**{stock_info['currentPrice']:,.2f} {stock_info['currency']}** || {(price_change):,.2f} ({(percent_change):,.2f}%) {updown}"
                    if stock_info['currentPrice'] is not None else 'None',
                    inline=False)

    # Post-Market Price
    if not isOpen and isnaq:
        if float(data['postMarketChange']) > 0:
            embed.colour = discord.Colour.green()
            updown = '<:yangbonghoro:>'
        elif float(data['postMarketChange']) < 0:
            embed.colour = discord.Colour.red()
            updown = '<:sale:>'
        else:
            embed.colour = discord.Colour.dark_blue()
            updown = ''

        embed.add_field(name=" ", value=" ", inline=False)
        embed.add_field(name="**Post-Market Price**", value=f"**{data['postMarketPrice']} {stock_info['currency']}** || {data['postMarketChange']} {data['postMarketChangePercent']} {updown}"
                        if data is not None else 'None',
                        inline=True)

    # Market Cap / Trading Volume
    embed.add_field(name=" ", value=" ", inline=False)
    embed.add_field(name="Market Cap", value=f"{stock_info['marketCap']:,} {stock_info['currency']}" if stock_info['marketCap'] is not None else 'None', inline=True)
    embed.add_field(name="Trading Volume", value=f"{stock_info['volume']:,}" if stock_info['volume'] is not None else 'None', inline=True)

이런 식으로 임베드를 만들어 넘겨주게 된다.

 

사용된 기타 함수들은 아래와 같다.

# Function to create Red-Black Tree from CSV file and save it to a JSON file
def create_rbt_from_csv_and_save(rbtree, csvPath, jsonPath):
    rbtree.insert_from_csv(csvPath)
    rbtree.save_to_json(jsonPath)

def check_file_exist():
    try:
        with open(jsonPath, "r") as json_file:
            Logger.debug("rbt.json exist")
            return True
    except FileNotFoundError:
        Logger.debug("rbt.json not exist")
        return False

def is_nasdaq_stocks(ticker):
        
    # Check rbt.search_by_key(ticker) is empty
    if not rbt.search_by_key(ticker):
        Logger.debug(f"{ticker} is not NASDAQ stock")
        return False
    
    return True

def is_nasdaq_open():
    # Get the NASDAQ market calendar
    nasdaq = mcal.get_calendar('NASDAQ')

    # Get the current date and time in Eastern Time (since NASDAQ operates in ET)
    now = datetime.datetime.now(pytz.timezone('US/Eastern'))

    # Get today's market schedule
    schedule = nasdaq.schedule(start_date=now.date(), end_date=now.date())

    # Check if the schedule is empty (i.e., the market is closed today)
    if schedule.empty:
        return False

    # Check if the current time is within today's market open hours
    return schedule.iloc[0].market_open <= now <= schedule.iloc[0].market_close

 

나스닥 상장주인지 확인하는 부분에 Red-Black Tree를 활용했다.

Nasdaq Stock Screener가 제공하는 csv파일을 통해 트리 형식으로 만들고 직렬화해 json 포맷으로 저장한다.

그걸 활용하는 식.

 

3-4. Red-Black Tree

트리 구현이 거기서 거기니 만큼 여기 다 푸는 건 너무 길어져서 그렇고...

검색을 조금 바꿨다.

 

def _search_by_value_recursive(self, node, value):
    if node == self.NIL_LEAF:
        return []
    result = []
    if value in node.value.lower():
        result.append(node.key)
    result += self._search_by_value_recursive(node.left, value)
    result += self._search_by_value_recursive(node.right, value)
    return result

검색어를 입력하면 그 키워드를 포함하는 모든 키를 리턴하게 했다.

검색 시 종목 별 가중치를 둬서 높은 가중치를 가진 종목이 상위에 뜨게 하면 될 듯?

VM 성능 때문에, 구상은 했지만 구현은 하지 않고 있다.

여하튼 여기서 키가 검색이 되면 상장돼 있는 것으로 간주해 판별한다.

 

이런 과정을 거쳐서 아래와 같은 결과를 만들어 내는 것이다.

 

4. 하지만 시간이...

성능이 안 좋으므로 셀레니움을 사용할 때 시간이 꽤 걸리게 된다.

최우선 해결과제라고 할 수 있겠다.