# 퀀티와이즈 재무제표 데이터 SQL 전송
import pandas as pd
from openpyxl import load_workbook
from sqlalchemy import create_engine
# 엑셀 파일 경로와 시트 이름 설정
file_name = 'fs_202312.xlsx'
sheet_name = 'value'
# 데이터 불러오기 (14행부터 시작, 필요한 열만 선택)
df = pd.read_excel(file_name, sheet_name=sheet_name, usecols="A,D,E,F,G,H", skiprows=13, header=None)
# 열 이름을 지정
df.columns = ['code', 'fs_op', 'fs_ni', 'fs_sales', 'fs_bps', 'fs_eps']
# D10 셀에서 period 값 가져오기
period = pd.read_excel(file_name, sheet_name=sheet_name, usecols="D", nrows=10, header=None).iloc[9, 0]
# period 값을 YYYY-MM 형식으로 변환 (문자열로 변환 후 슬라이싱)
period_str = str(period)
formatted_period = period_str[:4] + '-' + period_str[4:]
# period 열을 추가하고 변환된 period 값으로 채우기
df['period'] = formatted_period
# 데이터 프레임 확인
print(df)
# 각 데이터프레임 생성
df_fs_op = df[['period', 'code', 'fs_op']].rename(columns={'fs_op': 'value'})
df_fs_ni = df[['period', 'code', 'fs_ni']].rename(columns={'fs_ni': 'value'})
df_fs_sales = df[['period', 'code', 'fs_sales']].rename(columns={'fs_sales': 'value'})
df_fs_bps = df[['period', 'code', 'fs_bps']].rename(columns={'fs_bps': 'value'})
df_fs_eps = df[['period', 'code', 'fs_eps']].rename(columns={'fs_eps': 'value'})
# value 열에 결측치가 있는 행을 제거
df_fs_op = df_fs_op.dropna(subset=['value'])
df_fs_ni = df_fs_ni.dropna(subset=['value'])
df_fs_sales = df_fs_sales.dropna(subset=['value'])
df_fs_bps = df_fs_bps.dropna(subset=['value'])
df_fs_eps = df_fs_eps.dropna(subset=['value'])
# 결측치 제거 후 각 데이터프레임의 행 개수 확인
row_count_fs_op = len(df_fs_op)
row_count_fs_ni = len(df_fs_ni)
row_count_fs_sales = len(df_fs_sales)
row_count_fs_bps = len(df_fs_bps)
row_count_fs_eps = len(df_fs_eps)
print("df_fs_op 행 개수:", row_count_fs_op)
print("df_fs_ni 행 개수:", row_count_fs_ni)
print("df_fs_sales 행 개수:", row_count_fs_sales)
print("df_fs_bps 행 개수:", row_count_fs_bps)
print("df_fs_eps 행 개수:", row_count_fs_eps)
# # 분할된 데이터프레임 확인 (각 데이터프레임의 상위 5개 행을 확인)
# print("df_fs_op:\n", df_fs_op.head())
# print("df_fs_ni:\n", df_fs_ni.head())
# print("df_fs_sales:\n", df_fs_sales.head())
# print("df_fs_bps:\n", df_fs_bps.head())
# print("df_fs_eps:\n", df_fs_eps.head())
# 데이터 전송 경로 지정
db_url = 'mysql+mysqlconnector://motive:mtv1101@MOTIVE-WHWANG-02/motive'
# MySQL 연결 설정
db_connection = create_engine(db_url)
# 각 데이터프레임을 SQL로 전송
df_fs_op.to_sql(name='fs_op', con=db_connection, if_exists='append', index=False)
df_fs_ni.to_sql(name='fs_ni', con=db_connection, if_exists='append', index=False)
df_fs_sales.to_sql(name='fs_sales', con=db_connection, if_exists='append', index=False)
df_fs_bps.to_sql(name='fs_bps', con=db_connection, if_exists='append', index=False)
df_fs_eps.to_sql(name='fs_eps', con=db_connection, if_exists='append', index=False)
# 퀀티와이즈 컨센서스 데이터 SQL 전송
import pandas as pd
from openpyxl import load_workbook
from sqlalchemy import create_engine
# 엑셀 파일을 데이터프레임으로 가져와 롱폼으로 변형, SQL로 전송하는 함수
def cons_tp_op(file_path, sheet_names, table_name, db_url):
# 데이터를 저장할 빈 리스트
df_list = []
# 각 시트에서 데이터를 처리하는 함수
def process_sheet(sheet_name):
# openpyxl을 사용하여 엑셀 파일 로드
wb = load_workbook(file_path, data_only=True)
ws = wb[sheet_name]
# A열에서 데이터가 존재하는 마지막 행 찾기
last_row = ws.max_row
for row in range(ws.max_row, 0, -1):
if ws[f'A{row}'].value is not None:
last_row = row
break
# 8번째 행을 열 이름으로 가져오기
column_names = pd.read_excel(file_path, sheet_name=sheet_name, header=None, skiprows=7, nrows=1).iloc[0]
# 15번째 행부터 마지막 행까지 데이터를 불러옴
df = pd.read_excel(file_path, sheet_name=sheet_name, header=None, skiprows=14, nrows=last_row-14)
# 열 이름 설정
df.columns = column_names
df.columns.values[0] = "date"
# 변환 전 데이터프레임의 크기 출력 (행, 열)
print(f"Sheet: {sheet_name}, Before melt: {df.shape}")
# 롱폼으로 변환
df_long = pd.melt(df, id_vars=["date"], var_name="code", value_name=sheet_name)
# 변환 후 데이터프레임의 크기 출력 (행, 열)
print(f"Sheet: {sheet_name}, After melt: {df_long.shape}")
# 필요한 열만 반환 (date, code, sheet_name 열만 반환)
return df_long[["date", "code", sheet_name]]
# 각 시트에 대해 데이터를 처리하고 리스트에 저장
for sheet in sheet_names:
df_list.append(process_sheet(sheet))
# date, code를 기준으로 데이터프레임을 병합
df_final = df_list[0]
for df in df_list[1:]:
df_final = pd.merge(df_final, df, on=["date", "code"], how="outer")
# avg 열에 결측치(NaN)가 있는 행을 삭제
df_final_cleaned = df_final.dropna(subset=['avg'])
# 최종적으로 SQL로 전송할 데이터프레임의 행 개수 출력
print(f"Final cleaned DataFrame row count: {df_final_cleaned.shape[0]}")
# MySQL 연결 설정
db_connection = create_engine(db_url)
# df_final_cleaned를 지정된 테이블에 추가 (if_exists='append'는 데이터를 테이블에 추가하는 옵션)
df_final_cleaned.to_sql(name=table_name, con=db_connection, if_exists='append', index=False)
def cons_per(file_path, sheet_names, table_name, db_url):
# 데이터를 저장할 빈 리스트
df_list = []
# 각 시트에서 데이터를 처리하는 함수
def process_sheet(sheet_name):
# openpyxl을 사용하여 엑셀 파일 로드
wb = load_workbook(file_path, data_only=True)
ws = wb[sheet_name]
# A열에서 데이터가 존재하는 마지막 행 찾기
last_row = ws.max_row
for row in range(ws.max_row, 0, -1):
if ws[f'A{row}'].value is not None:
last_row = row
break
# 8번째 행을 열 이름으로 가져오기
column_names = pd.read_excel(file_path, sheet_name=sheet_name, header=None, skiprows=7, nrows=1).iloc[0]
# 15번째 행부터 마지막 행까지 데이터를 불러옴
df = pd.read_excel(file_path, sheet_name=sheet_name, header=None, skiprows=14, nrows=last_row-14)
# 열 이름 설정
df.columns = column_names
df.columns.values[0] = "date"
# 변환 전 데이터프레임의 크기 출력 (행, 열)
print(f"Sheet: {sheet_name}, Before melt: {df.shape}")
# 롱폼으로 변환
df_long = pd.melt(df, id_vars=["date"], var_name="code", value_name=sheet_name)
# 변환 후 데이터프레임의 크기 출력 (행, 열)
print(f"Sheet: {sheet_name}, After melt: {df_long.shape}")
# 필요한 열만 반환 (date, code, sheet_name 열만 반환)
return df_long[["date", "code", sheet_name]]
# 각 시트에 대해 데이터를 처리하고 리스트에 저장
for sheet in sheet_names:
df_list.append(process_sheet(sheet))
# 첫 번째 데이터프레임을 기준으로 다른 시트의 데이터프레임을 병합
df_final = df_list[0]
for df in df_list[1:]:
df_final = pd.merge(df_final, df, on=["date", "code"], how="outer")
# value 열에 결측치(NaN)가 있는 행을 삭제
df_final_cleaned = df_final.dropna(subset=['value'])
# 최종적으로 SQL로 전송할 데이터프레임의 행 개수 출력
print(f"Final cleaned DataFrame row count: {df_final_cleaned.shape[0]}")
# MySQL 연결 설정
db_connection = create_engine(db_url)
# df_final_cleaned를 지정된 테이블에 추가 (if_exists='append'는 데이터를 테이블에 추가하는 옵션)
df_final_cleaned.to_sql(name=table_name, con=db_connection, if_exists='append', index=False)
# 데이터 전송 경로 지정
db_url = 'mysql+mysqlconnector://motive:mtv1101@MOTIVE-WHWANG-02/motive'
# cons_tp 파일경로, 시트 지정
cons_tp_file_path = 'cons_tp_20240705_20240912.xlsx'
cons_tp_sheet_names = ['avg', 'high', 'low', 'num']
cons_tp_table_name = 'cons_tp'
# cons_op_ifq1 파일경로, 시트 지정
cons_op_ifq1_file_path = 'cons_op_ifq1_20240701_20240912.xlsx'
cons_op_ifq1_sheet_names = ['avg', 'high', 'low', 'std', 'num']
cons_op_ifq1_table_name = 'cons_op_ifq1'
# cons_op_ifq2 파일경로, 시트 지정
cons_op_ifq2_file_path = 'cons_op_ifq2_20240326_20240912.xlsx'
cons_op_ifq2_sheet_names = ['avg', 'high', 'low', 'num']
cons_op_ifq2_table_name = 'cons_op_ifq2'
# cons_op_ify1 파일경로, 시트 지정
cons_op_ify1_file_path = 'cons_op_ify1_20240701_20240912.xlsx'
cons_op_ify1_sheet_names = ['avg', 'high', 'low', 'num']
cons_op_ify1_table_name = 'cons_op_ify1'
# cons_per_f12m 파일경로, 시트 지정
cons_per_f12m_file_path = 'cons_per_f12m_20230703_20240912.xlsx'
cons_per_f12m_sheet_names = ['value']
cons_per_f12m_table_name = 'cons_per_f12m'
# 함수 실행
# cons_tp_op(cons_tp_file_path, cons_tp_sheet_names, cons_tp_table_name, db_url)
# cons_tp_op(cons_op_ifq1_file_path, cons_op_ifq1_sheet_names, cons_op_ifq1_table_name, db_url)
# cons_tp_op(cons_op_ifq2_file_path, cons_op_ifq2_sheet_names, cons_op_ifq2_table_name, db_url)
# cons_tp_op(cons_op_ify1_file_path, cons_op_ify1_sheet_names, cons_op_ify1_table_name, db_url)
# cons_per(cons_per_f12m_file_path, cons_per_f12m_sheet_names, cons_per_f12m_table_name, db_url)
'Python' 카테고리의 다른 글
| 주주 현황 크롤링_fnguide (0) | 2024.10.03 |
|---|---|
| Qunat전략_무상증자 (3) | 2024.09.25 |
| SEIBro 채권정보 수집 (iframe 크롤링) (1) | 2024.07.03 |
| 일별 종가 데이터로 월별 수익률 구하기 (2) | 2024.07.03 |
| 종가시계열 시각화 컬러맵 (중앙값 기준 정규화) (1) | 2024.04.27 |