model/src/data/financial_data.py

"""
재무제표 데이터 처리 관련 함수
"""

import pandas as pd
import numpy as np
import requests
import warnings
from bs4 import BeautifulSoup
from statsmodels.tsa.holtwinters import ExponentialSmoothing

warnings.filterwarnings("ignore", message="Optimization failed to converge")

def scrape_financial_statement(ticker, statement_type):
    """
    특정 종류의 재무제표를 스크래핑하는 함수
    """
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36'
    }
    
    # URL 매핑
    url_paths = {
        'income': '',
        'ratios': 'ratios/',
        'balance-sheet': 'balance-sheet/',
        'cash-flow-statement': 'cash-flow-statement/'
    }
    
    statement_names = {
        'income': '수익계산서',
        'ratios': '재무비율',
        'balance-sheet': '대차대조표',
        'cash-flow-statement': '현금흐름표'
    }
    
    try:
        url = f"https://stockanalysis.com/stocks/{ticker}/financials/{url_paths[statement_type]}?p=quarterly"
        response = requests.get(url, headers=headers)
        print(f"{statement_names[statement_type]} 상태코드: {response.status_code}")
        
        soup = BeautifulSoup(response.content, 'html.parser')
        element_tables = soup.select("table[data-test='financials']")
        
        if not element_tables:
            print(f"{ticker}: {statement_names[statement_type]} 테이블을 찾을 수 없습니다.")
            return None
        
        df = pd.read_html(str(element_tables))[0]
        
        # 컬럼이 MultiIndex인지 확인
        if isinstance(df.columns, pd.MultiIndex):
            print(f"{ticker}: MultiIndex {statement_names[statement_type]} 처리")
            date_cols = df.columns.get_level_values(1)[1:]  # Period Ending 값들
            df = df.droplevel(0, axis=1)  # 첫번째 레벨 제거
            df.columns = [df.columns[0]] + list(date_cols)
            result_df = df.set_index(df.columns[0]).transpose()
        else:
            date_col = df.columns[0]
            result_df = df.set_index(date_col).transpose()
        
        result_df.index.name = "Date"
        
        # 첫 행 제외
        if statement_type == 'ratios':
            result_df = result_df.iloc[1:-1, :]  # 첫 행과 마지막 행 제외
        else:
            result_df = result_df.iloc[:-1, :]  # 마지막 행만 제외
        
        return result_df
    
    except Exception as e:
        print(f"{ticker} {statement_names[statement_type]} 스크래핑 오류: {e}")
        return None

def convert_to_numeric(df):
    """
    DataFrame의 모든 열을 숫자형으로 변환
    """
    for column in df.columns:
        if df[column].dtype == 'object':
            # 음수값 처리 (예: '-123' -> -123)
            df[column] = df[column].apply(
                lambda x: float(str(x).replace('-', '')) * -1 
                if isinstance(x, str) and '-' in x and x.replace('-', '').replace('.', '').isdigit() 
                else x
            )
            
            # 백분율 처리 (예: '12%' -> 0.12)
            if df[column].dtype == 'object':
                df[column] = df[column].apply(
                    lambda x: float(str(x).replace('%', '')) / 100 
                    if isinstance(x, str) and '%' in x 
                    else x
                )
            
            # 최종 숫자 변환
            df[column] = pd.to_numeric(df[column], errors='coerce')
    
    return df

def convert_date_format(date_str):
    """
    날짜 문자열을 표준 형식으로 변환
    """
    try:
        # 미래 날짜 처리 개선
        if isinstance(date_str, str) and "'" in date_str and len(date_str.split()) >= 4:
            # 미래 데이터 감지 시 None 반환
            if any(future_marker in date_str for future_marker in ["'24", "2024"]):
                return None
                
            # 과거 데이터는 정상 처리
            parts = date_str.split()
            month_part = parts[-3]
            day_part = parts[-2].replace(',', '')
            year_part = parts[-1]
            
            month_dict = {
                'Jan': 1, 'Feb': 2, 'Mar': 3, 'Apr': 4, 'May': 5, 'Jun': 6,
                'Jul': 7, 'Aug': 8, 'Sep': 9, 'Oct': 10, 'Nov': 11, 'Dec': 12
            }
            
            month = month_dict.get(month_part, 1)
            day = int(day_part)
            year = int(year_part)
            
            return f"{year}-{month:02d}-{day:02d}"
                
        # 기타 형식 처리
        return date_str
            
    except Exception as e:
        print(f"날짜 변환 오류: {e} - '{date_str}'")
        return None

def process_index_dates(df):
    """
    데이터프레임의 인덱스 날짜를 변환
    """
    new_index = [convert_date_format(idx) for idx in df.index]
    df['_temp_date'] = new_index
    df = df[df['_temp_date'].notna()]
    
    if df.empty:
        return None
    
    df.index = df['_temp_date']
    df = df.drop(columns=['_temp_date'])
    return df

def interpolate_and_forecast(df, end_date):
    """
    재무 데이터를 일별로 보간하고 필요시 미래 데이터 예측
    """
    # 날짜 인덱스를 datetime으로 변환
    df.index = pd.to_datetime(df.index)
    
    # 일별 데이터로 변환 및 보간
    daily_df = df.resample('D').asfreq()
    for column in daily_df.columns:
        daily_df[column] = daily_df[column].interpolate(method='linear')
    
    # 예측 데이터 생성
    end_date = pd.to_datetime(end_date)
    forecast_steps = (end_date - daily_df.index[-1]).days
    
    if forecast_steps > 0:
        print(f"예측 시작: {forecast_steps}일")
        date_range = pd.date_range(daily_df.index[-1] + pd.Timedelta(days=1), end_date)
        
        # 예측값을 사전에 먼저 모음
        forecasts = {}
        
        for column in daily_df.columns:
            try:
                model = ExponentialSmoothing(
                    daily_df[column], trend='add', seasonal=None, seasonal_periods=4
                ).fit() 
                forecast = model.forecast(steps=forecast_steps)
                forecasts[column] = forecast
            except Exception as e:
                print(f"{column} 예측 실패: {e}")
                forecasts[column] = np.full(forecast_steps, np.nan)
        
        # 한 번에 DataFrame 생성
        forecast_df = pd.DataFrame(forecasts, index=date_range)
        daily_df = pd.concat([daily_df, forecast_df])
    
    # 결측치가 있는 열 제거
    daily_df = daily_df.dropna(axis=1, how='any')
    
    return daily_df

def process_financial_data(ticker, all_data, stock_end_date):
    """
    재무제표 데이터를 처리하는 메인 함수
    """
    try:
        print(f"===== {ticker} 재무데이터 처리 시작 =====")
        
        # 각 재무제표 스크래핑
        FS_Income = scrape_financial_statement(ticker, 'income')
        FS_Ratio = scrape_financial_statement(ticker, 'ratios')
        FS_Balance = scrape_financial_statement(ticker, 'balance-sheet')
        FS_Cash = scrape_financial_statement(ticker, 'cash-flow-statement')
        
        # 스크래핑 실패 확인
        if any(fs is None for fs in [FS_Income, FS_Ratio, FS_Balance, FS_Cash]):
            print(f"{ticker}: 일부 재무제표 데이터를 가져오지 못했습니다.")
            return None
        
        # 재무제표 데이터를 숫자로 변환
        FS_Income = convert_to_numeric(FS_Income)
        FS_Ratio = convert_to_numeric(FS_Ratio)
        FS_Balance = convert_to_numeric(FS_Balance)
        FS_Cash = convert_to_numeric(FS_Cash)
        
        # 날짜 인덱스 처리
        FS_Income = process_index_dates(FS_Income)
        FS_Ratio = process_index_dates(FS_Ratio)
        FS_Balance = process_index_dates(FS_Balance)
        FS_Cash = process_index_dates(FS_Cash)
        
        # 날짜 변환 실패 확인
        if any(fs is None for fs in [FS_Income, FS_Ratio, FS_Balance, FS_Cash]):
            print(f"{ticker}: 날짜 변환 후 유효한 데이터가 없습니다.")
            return None
        
        # ROE 계산
        try:
            if 'Net Income' in FS_Income.columns and 'Shareholders\' Equity' in FS_Balance.columns:
                FS_Ratio['ROE'] = FS_Income['Net Income'] / FS_Balance['Shareholders\' Equity']
        except Exception as e:
            print(f"ROE 계산 오류: {e}")
        
        # 모든 재무제표 데이터 병합
        FS_Summary = pd.concat([FS_Income, FS_Balance, FS_Ratio, FS_Cash], axis=1)
        
        # 중복 컬럼 제거
        duplicated_columns = FS_Summary.columns[FS_Summary.columns.duplicated()].unique()
        if len(duplicated_columns) > 0:
            print(f"{ticker} 중복 컬럼 제거: {duplicated_columns}")
            FS_Summary = FS_Summary.loc[:, ~FS_Summary.columns.duplicated()]
        
        # 보간 및 예측
        daily_FS_Summary = interpolate_and_forecast(FS_Summary, stock_end_date)
        
        if daily_FS_Summary.empty:
            print(f"{ticker}: 유효한 일별 재무 데이터가 없습니다")
            return None

        # 주가 데이터 병합
        if ticker in all_data and 'Close' in all_data[ticker].columns:
            close_df = pd.DataFrame(all_data[ticker]['Close'])
            close_df.columns = ['Close']
            
            # 재무 데이터와 주가 데이터 병합
            daily_FS_Summary = daily_FS_Summary.merge(
                close_df, left_index=True, right_index=True, how='inner'
            )
            
            if daily_FS_Summary.empty:
                print(f"{ticker}: 주가 데이터와 병합 후 데이터가 없습니다")
                return None
        else:
            print(f"{ticker}: Close 데이터를 찾을 수 없습니다")
            return None
                
        print(f"{ticker} 재무 데이터 처리 완료: {daily_FS_Summary.shape}")
        return daily_FS_Summary
        
    except Exception as e:
        print(f"{ticker} 처리 중 오류 발생: {e}")
        import traceback
        traceback.print_exc()
        return None