DART 자사주 취득, 무상증자 공시 알림

import requests

import smtplib

from email.mime.multipart import MIMEMultipart

from email.mime.text import MIMEText

import csv

import pandas as pd

from pykrx import stock

from datetime import datetime, timedelta

import schedule

import time

import telegram

import asyncio

# 이전 실행에서의 최종 데이터셋을 저장할 전역 변수

df_old = None

count = 1

# 오늘 날짜를 yyyyMMdd 형식으로 저장

today_date = datetime.now().strftime("%Y%m%d")

# 어제 날짜

yesterday_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")

# 텔레그램 메세지 전송

async def send_message(a):

    token = " "

    bot = telegram.Bot(token)

    await bot.send_message(chat_id=" ", text=a)

list_result = []

# DART API 정보

api_key = ' '

url_1 = 'https://opendart.fss.or.kr/api/list.json'

def main():

    try:

        global df_old

        global count

        # DART에서 공시 정보 조회

        params = {

            'crtfc_key': api_key,

            'pblntf_ty' : 'B',

            'sort' : 'date',

            'sort_mth' : 'desc',

            'bgn_de': yesterday_date,  # 시작 날짜

            'end_de': today_date,  # 종료 날짜

            'page_no' : '5',

            'page_count' : '100'

        }

        response = requests.get(url_1, params=params)

        data_1 = response.json()

        # list 키의 값을 DataFrame으로 변환

        df = pd.DataFrame(data_1['list'])

        # 필요한 열만 선택

        df = df[['corp_code', 'corp_name', 'stock_code', 'rcept_dt', 'report_nm']]

        # '자기주식취득결정' or '자기추식취득신탁계약체결결정' or '무상증자결정'을 포함하는 행만 필터링

        df_filtered = df[df['report_nm'].str.contains('자기주식취득결정|자기주식취득신탁계약체결결정|무상증자결정')]

        # 특정 텍스트를 포함하는 행 제외

        exclude_texts = [

            '기재정정', '첨부정정', '첨부추가', '변경등록',

            '연장결정', '발행조건확정', '정정명령부과', '정정제출요구'

        ]

        pattern = '|'.join(exclude_texts)  # OR 조건을 위한 정규식 패턴 생성

        df_final = df_filtered[~df_filtered['report_nm'].str.contains(pattern)]

       

        # 새로운 데이터가 있는지 확인, 새로운 데이터는 new_entries에 저장

        if df_old is not None:

            new_entries = pd.concat([df_final, df_old, df_old]).drop_duplicates(keep=False)

           

            if not new_entries.empty:

                # 데이터프레임 가독성 좋은 서식으로 문자변수로 바꾸기

               

                # 행 길이 측정

                rows_length = len(new_entries)

               

                # 서식 변경 (행 갯수 만큼)

                for i in range(0, rows_length):

                    corp_code = new_entries.iloc[i, 0]

                    corp_name = new_entries.iloc[i, 1]

                    stock_code = new_entries.iloc[i, 2]

                    rcept_dt = new_entries.iloc[i, 3]

                    report_nm = new_entries.iloc[i, 4]

                   

                    formatted_str = f"corp_code : {corp_code}\ncorp_name : {corp_name}\nstock_code : {stock_code}\nrcept_dt : {rcept_dt}\nreport_nm : {report_nm}"

                   

                    # 새로운 데이터를 텔레그램으로 전송

                    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

                    asyncio.run(send_message(formatted_str))

                   

                print("새로운 데이터가 추가되었습니다:\n", formatted_str)

               

            else:

                print(f"새로운 데이터가 없습니다 : {count}")

               

        # df_old 업데이트

        df_old = df_final.copy()

        count = count + 1

       

    except Exception as e:

        print(f'error : {count}')

# 실행 주기 설정

schedule.every(5).seconds.do(main)

# 파이썬 파일 종료 방지

while True:

    schedule.run_pending()

    time.sleep(1)