스트림릿 웹앱으로 KRX 에서 필요한 주식데이터 보기

작성된 코드import streamlit as st
import pandas as pd
from datetime import datetime, timedelta
import os
import sys
import subprocess

# 페이지 설정 (반드시 첫 번째 Streamlit 명령어여야 함)
st.set_page_config(
    page_title="KRX 주식 데이터 분석",
    page_icon="📈",
    layout="wide"
)

# 필요한 패키지 설치 확인 및 설치
def install_package(package):
    try:
        __import__(package)
    except ImportError:
        st.info(f"{package} 패키지를 설치합니다...")
        subprocess.check_call([sys.executable, "-m", "pip", "install", package])
        st.success(f"{package} 패키지가 설치되었습니다.")

# 필요한 패키지 설치
install_package("pykrx")
install_package("matplotlib")
install_package("numpy")
install_package("requests")
install_package("beautifulsoup4")
install_package("python-dateutil")

# pykrx 모듈 import
from pykrx import stock

from data_collector import KRXDataCollector
from data_processor import DataProcessor
from utils import format_date, get_last_business_day, validate_inputs

# 제목
st.title("KRX 주식 데이터 분석")

# 사이드바 - 입력 파라미터
st.sidebar.header("검색 조건")

# 날짜 선택 (오늘 날짜로 기본 설정)
today = datetime.now().date()
selected_date = st.sidebar.date_input(
    "날짜 선택",
    value=today,
    max_value=today,
    min_value=datetime(2020, 1, 1).date()
)

# 최소 종가 입력 (정수만)
min_price = st.sidebar.number_input(
    "최소 종가",
    min_value=0,
    value=1000,
    step=100,
    format="%d"  # 정수 형식으로 표시
)

# 최소 거래량 입력
min_volume = st.sidebar.number_input(
    "최소 거래량",
    min_value=0,
    value=10000,
    step=1000
)

# 최소 등락률 입력 (정수만)
min_change_rate = st.sidebar.number_input(
    "최소 등락률",
    min_value=0,
    value=1,
    step=1,
    format="%d"  # 정수 형식으로 표시
)

# 데이터 수집 및 처리
if st.sidebar.button("데이터 조회"):
    # 입력값 검증
    is_valid, error_message = validate_inputs(min_price, min_volume, min_change_rate)
    
    if not is_valid:
        st.error(error_message)
    else:
        with st.spinner("데이터를 수집하고 있습니다..."):
            # 데이터 수집
            collector = KRXDataCollector()
            date_str = format_date(selected_date)
            df = collector.get_stock_data(date_str)
            
            if df.empty:
                st.error("데이터를 수집하는 중 오류가 발생했습니다.")
            else:
                # 데이터 필터링
                processor = DataProcessor()
                filtered_df = processor.filter_data(df, min_price, min_volume, min_change_rate)
                
                if filtered_df.empty:
                    st.warning("조건에 맞는 데이터가 없습니다.")
                else:
                    # 결과 표시
                    st.subheader("필터링된 결과")
                    st.write(f"검색 결과: {len(filtered_df)}개 종목 (등락률 > {min_change_rate}%인 종목 중에서 검색)")
                    st.dataframe(
                        filtered_df,
                        use_container_width=True,
                        hide_index=True
                    )
                    
                    # CSV 다운로드 버튼
                    csv = filtered_df.to_csv(index=False, encoding='utf-8-sig')
                    st.download_button(
                        label="CSV 다운로드",
                        data=csv,
                        file_name=f"krx_data_{date_str}.csv",
                        mime="text/csv"
                    )

# 사용 방법 안내
with st.expander("사용 방법"):
    st.markdown("""
    1. 왼쪽 사이드바에서 날짜를 선택합니다.
    2. 최소 종가를 입력합니다.
    3. 최소 거래량을 입력합니다.
    4. 최소 등락률을 입력합니다.
    5. '데이터 조회' 버튼을 클릭합니다.
    6. 필터링된 결과가 표시됩니다.
    7. 'CSV 다운로드' 버튼을 클릭하여 데이터를 저장합니다.
    """) 

import pandas as pd
from datetime import datetime
from pykrx import stock
import concurrent.futures
import os
import pickle
from pathlib import Path

class KRXDataCollector:
    def __init__(self):
        self.cache_dir = Path("data/cache")
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        
    def get_stock_data(self, date_str):
        """
        pykrx 라이브러리를 사용하여 KRX 웹사이트에서 주식 데이터를 수집합니다.
        캐싱을 사용하여 이미 수집한 데이터는 재사용합니다.
        
        Args:
            date_str (str): 'YYYYMMDD' 형식의 날짜 문자열
            
        Returns:
            pd.DataFrame: 수집된 주식 데이터
        """
        # 캐시 파일 경로
        cache_file = self.cache_dir / f"krx_data_{date_str}.pkl"
        
        # 캐시된 데이터가 있으면 로드
        if cache_file.exists():
            try:
                with open(cache_file, 'rb') as f:
                    return pickle.load(f)
            except Exception as e:
                print(f"캐시 로드 중 오류 발생: {str(e)}")
        
        try:
            # 코스피 종목 리스트 가져오기
            kospi_tickers = stock.get_market_ticker_list(market="KOSPI")
            kosdaq_tickers = stock.get_market_ticker_list(market="KOSDAQ")
            
            # 병렬 처리를 위한 함수
            def fetch_stock_data(ticker, market):
                try:
                    df = stock.get_market_ohlcv_by_date(date_str, date_str, ticker)
                    if not df.empty:
                        df['종목코드'] = ticker
                        df['종목명'] = stock.get_market_ticker_name(ticker)
                        df['시장구분'] = market
                        return df
                except:
                    pass
                return None
            
            # 병렬 처리로 데이터 수집
            all_data = []
            
            # 코스피 데이터 수집
            with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
                future_to_ticker = {executor.submit(fetch_stock_data, ticker, 'KOSPI'): ticker for ticker in kospi_tickers}
                for future in concurrent.futures.as_completed(future_to_ticker):
                    result = future.result()
                    if result is not None:
                        all_data.append(result)
            
            # 코스닥 데이터 수집
            with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
                future_to_ticker = {executor.submit(fetch_stock_data, ticker, 'KOSDAQ'): ticker for ticker in kosdaq_tickers}
                for future in concurrent.futures.as_completed(future_to_ticker):
                    result = future.result()
                    if result is not None:
                        all_data.append(result)
            
            # 데이터 합치기
            if all_data:
                combined_data = pd.concat(all_data)
                
                # 필요한 컬럼만 선택
                result_df = combined_data[['종목코드', '종목명', '시장구분', '종가', '거래량']].copy()
                
                # 등락률 계산
                result_df['등락률'] = ((combined_data['종가'] - combined_data['시가']) / combined_data['시가'] * 100).round(2)
                
                # 결과 캐싱
                try:
                    with open(cache_file, 'wb') as f:
                        pickle.dump(result_df, f)
                except Exception as e:
                    print(f"캐시 저장 중 오류 발생: {str(e)}")
                
                return result_df
            else:
                print("수집된 데이터가 없습니다.")
                return pd.DataFrame()
            
        except Exception as e:
            print(f"데이터 수집 중 오류 발생: {str(e)}")
            return pd.DataFrame() 

import pandas as pd
import os

class DataProcessor:
    @staticmethod
    def filter_data(df, min_price, min_volume, min_change_rate):
        """
        주어진 조건에 따라 데이터를 필터링합니다.
        
        Args:
            df (pd.DataFrame): 필터링할 데이터프레임
            min_price (float): 최소 종가
            min_volume (float): 최소 거래량
            min_change_rate (float): 최소 등락률
            
        Returns:
            pd.DataFrame: 필터링된 데이터프레임
        """
        try:
            # 등락률이 양수이고 최소 등락률 이상인 종목만 선택
            filtered_df = df[
                (df['등락률'] > 0) & 
                (df['등락률'] >= min_change_rate) &
                (df['종가'] >= min_price) & 
                (df['거래량'] >= min_volume)
            ]
            
            # 등락률 기준으로 내림차순 정렬
            filtered_df = filtered_df.sort_values('등락률', ascending=False)
            
            return filtered_df
            
        except Exception as e:
            print(f"데이터 필터링 중 오류 발생: {str(e)}")
            return pd.DataFrame()
    
    @staticmethod
    def save_to_csv(df, date_str):
        """
        데이터프레임을 CSV 파일로 저장합니다.
        
        Args:
            df (pd.DataFrame): 저장할 데이터프레임
            date_str (str): 'YYYYMMDD' 형식의 날짜 문자열
            
        Returns:
            str: 저장된 파일 경로
        """
        try:
            # 저장 경로 설정
            save_dir = r"C:\AI\Cursor\Stock"
            
            # 디렉토리가 없으면 생성
            if not os.path.exists(save_dir):
                os.makedirs(save_dir)
            
            filename = os.path.join(save_dir, f"krx_data_{date_str}.csv")
            df.to_csv(filename, index=False, encoding='utf-8-sig')
            return filename
            
        except Exception as e:
            print(f"CSV 파일 저장 중 오류 발생: {str(e)}")
            return None 

from datetime import datetime, timedelta

def format_date(date):
    """
    datetime 객체를 'YYYYMMDD' 형식의 문자열로 변환합니다.
    
    Args:
        date (datetime): 변환할 날짜
        
    Returns:
        str: 'YYYYMMDD' 형식의 날짜 문자열
    """
    return date.strftime('%Y%m%d')

def get_last_business_day():
    """
    마지막 거래일을 반환합니다.
    
    Returns:
        datetime: 마지막 거래일
    """
    today = datetime.now()
    last_day = today - timedelta(days=1)
    
    # 주말인 경우 금요일로 이동
    while last_day.weekday() >= 5:  # 5: 토요일, 6: 일요일
        last_day = last_day - timedelta(days=1)
        
    return last_day

def validate_inputs(min_price, min_volume, min_change_rate):
    """
    사용자 입력값을 검증합니다.
    
    Args:
        min_price (float): 최소 종가
        min_volume (float): 최소 거래량
        min_change_rate (float): 최소 등락률
        
    Returns:
        tuple: (is_valid, error_message)
    """
    if min_price < 0:
        return False, "최소 종가는 0 이상이어야 합니다."
    
    if not float(min_price).is_integer():
        return False, "최소 종가는 정수여야 합니다."
    
    if min_volume < 0:
        return False, "최소 거래량은 0 이상이어야 합니다."
        
    if min_change_rate < 0:
        return False, "최소 등락률은 0 이상이어야 합니다."
        
    if not float(min_change_rate).is_integer():
        return False, "최소 등락률은 정수여야 합니다."
        
    return True, ""