data-pipeline/update.py at main · React-4/data-pipeline · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
from data_collectors.currentPrice import fetch_sector_data
from connector import redis_connector, mysql_connector
from data_collectors import hantwo_api_topN
from data_collectors.disclosure_chatgpt_api import process_disclosures, process_disclosures2
from data_collectors.disclosure_dart_api import fetch_dart_filings
from data_collectors.price_crawler import stock_price_crawler
import pandas as pd
from datetime import datetime
import schedule
import time

# MySQL 및 Redis 연결 객체
mc = mysql_connector
redis_client = redis_connector.get_redis_client()

def get_mysql_connection():
    """
    MySQL 연결 확인 및 재연결 함수.
    """
    global mc
    try:
        # MySQL 연결 상태 확인
        mc.ping(reconnect=True)
    except Exception as e:
        print(f"MySQL 연결 재설정 중...: {e}")
        mc = mysql_connector  # 재연결 수행
    return mc


def get_redis_connection():
    """
    Redis 연결 확인 및 재연결 함수.
    """
    global redis_client
    try:
        # Redis 연결 상태 확인
        redis_client.ping()
    except Exception as e:
        print(f"Redis 연결 재설정 중...: {e}")
        redis_client = redis_connector.get_redis_client()  # 재연결 수행
    return redis_client


# 월~금 하루에 한번 오후 6시 실행
def update_day():
    current_time = datetime.now()
    print("start_day",current_time)
    mc = get_mysql_connection()
    stockInfo_df = mc.read_table_to_dataframe("stock")
    days_price_df = stock_price_crawler(stockInfo_df, "days", 1)
    mc.upload_dataframe_to_mysql(days_price_df, "stock_price_day", "append")


# 매주 금요일 6시에 한번 실행
def update_weeks():
    current_time = datetime.now()
    print("start_week",current_time)
    mc = get_mysql_connection()
    stockInfo_df = mc.read_table_to_dataframe("stock")
    weeks_price_df = stock_price_crawler(stockInfo_df, "weeks", 1)
    mc.upload_dataframe_to_mysql(weeks_price_df, "stock_price_week", "append")


# 매달 1일 한번 실행
def update_months():
    current_time = datetime.now()
    print("start_month",current_time)
    if datetime.now().day != 1:
        return

    mc = get_mysql_connection()
    stockInfo_df = mc.read_table_to_dataframe("stock")
    months_price_df = stock_price_crawler(stockInfo_df, "months", 1)
    mc.upload_dataframe_to_mysql(months_price_df, "stock_price_month", "append")


# 매일 10분에 한번씩 공시 크롤링
previous_row_count = 0


def reset_previous_row_count():
    global previous_row_count
    previous_row_count = 0
    # print(f"{datetime.now()} - previous_row_count 초기화 완료.")


def update_10m():
    current_time = datetime.now()
    if current_time.weekday() < 5 and 9 <= current_time.hour < 18:
        print("10분 간격 작업 실행 중...")
        print(current_time)

    global previous_row_count
    mc = get_mysql_connection()
    stockInfo_df = mc.read_table_to_dataframe("stock")
    today = datetime.today().strftime('%Y%m%d')
    dart_df = fetch_dart_filings(today, today, corp_cls='Y', page_count=25)
    current_row_count = len(dart_df)

    if current_row_count > previous_row_count:
        new_rows = dart_df.iloc[previous_row_count:]
        # print(f"New rows detected: {len(new_rows)}")
        previous_row_count = current_row_count
        df = process_disclosures2(new_rows, stockInfo_df)
        mc.upload_dataframe_to_mysql(df, "announcement", "append")
    else:
        print("No new rows detected.")


# 매 1분마다 실행
def update_1m():
    current_time = datetime.now()
    if current_time.weekday() < 5 and 9 <= current_time.hour < 18:
        print("1분 간격 작업 실행 중...")
        print(current_time)
    else:
        return


    redis_client = get_redis_connection()
    print("실시간 순위 데이터 수집 및 Redis 저장 시작...")
    ht = hantwo_api_topN

    # 상승률, 하락률, 거래량, 거래대금 순위 데이터 수집
    df_up = ht.fetch_change_rate(2)
    time.sleep(1)
    df_down = ht.fetch_change_rate(3)
    time.sleep(1)
    df_volume = ht.fetch_volume_or_transaction_rank("volume")
    time.sleep(1)
    df_transaction = ht.fetch_volume_or_transaction_rank("transaction")
    time.sleep(1)

    # Redis에 저장
    redis_connector.save_df_to_redis_as_nested_json(redis_client, df_up, "상승률순위")
    redis_connector.save_df_to_redis_as_nested_json(redis_client, df_down, "하락률순위")
    redis_connector.save_df_to_redis_as_nested_json(redis_client, df_volume, "거래량순위")
    redis_connector.save_df_to_redis_as_nested_json(redis_client, df_transaction, "거래대금순위")

    # 5. KOSPI/KOSDAQ 현재가 데이터 수집 및 Redis 저장
    print("KOSPI/KOSDAQ 현재가 데이터 수집 및 Redis 저장 시작...")
    kospi_current_price = fetch_sector_data('KOSPI')
    kosdaq_current_price = fetch_sector_data('KOSDAQ')

    # Redis에 저장
    redis_connector.save_stocks_to_redis(redis_client, kospi_current_price)
    redis_connector.save_stocks_to_redis(redis_client, kosdaq_current_price)
    print("KOSPI/KOSDAQ 현재가 데이터 저장 완료.")

    print("실시간 순위 데이터 저장 완료.")


# 작업 스케줄링
def main():
    schedule.every().monday.at("18:00").do(update_day)
    schedule.every().tuesday.at("18:00").do(update_day)
    schedule.every().wednesday.at("18:00").do(update_day)
    schedule.every().thursday.at("18:00").do(update_day)
    schedule.every().friday.at("18:00").do(update_day)
    schedule.every().day.at("00:00").do(reset_previous_row_count)
    schedule.every().friday.at("18:00").do(update_weeks)
    schedule.every().friday.at("18:30").do(update_months)
    schedule.every(10).minutes.do(update_10m)
    schedule.every(1).minutes.do(update_1m)

    print("스케줄링 시작...")
    while True:
        schedule.run_pending()
        time.sleep(1)

if __name__ == '__main__':
    main()
    # update_weeks()
    # update_months()
    # update_10m()
    # update_1m()