""" 재무제표 데이터 처리 관련 함수 """ 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