Airflow로 ChatGpt API 호출하기

BAO.DE·2025년 9월 22일

airflow chatGPT

Apache Airflow

목록 보기

15/20

API 사용을 위한 SecKey 발급

발급받은 키는 잘 보관해야한다.
창닫으면 사라짐!

Request

거금 10$를 충전해서 테스트를 진행

role , content를 지정해주고
assistant로 부터 받은 json 데이터를 파싱해주었다.

import os
import requests
import json

# 1) API 키 가져오기 (환경변수로 설정하는 것이 안전)
secret_key =  ' '
if not secret_key:
    raise ValueError("환경변수 OPENAI_API_KEY가 설정되지 않았습니다.")

url = "https://api.openai.com/v1/chat/completions"  # 올바른 엔드포인트

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {secret_key}"
}

data = {
    "model": "gpt-3.5-turbo",
    "messages": [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Airflow PythonOperator를 이용해서 sample dag 만들어줘"}
    ],
    "temperature": 0.7,
    "max_tokens": 200  # 응답의 최대 길이를 조절하고 싶으면 추가
}

response = requests.post(url, headers=headers, json=data)

if response.status_code == 200:
    result = response.json()
    # assistant의 응답 내용 추출
    assistant_message = result["choices"][0]["message"]["content"]
    print("GPT 응답:", assistant_message)
else:
    print("오류 발생:", response.status_code, response.text)

Response

response 결과에 실제 코드외에 한글설명등은 제외할 필요가 있어보인다.

비트코인 정보 요약 DAG

비트코인 트렌드에 민감하게 대응해야하는 비즈니스를 운영한다고 가정해보자.

가격 동향을 매일 뉴스 스크랩하는 것보다 daily task로 바이낸스를 통해서price 정보를 읽고 gpt로 요약후 EmailOperator를 통해서 메일을 받는다면? 그런 수고로움이 사라지고 해당 정보를 전사 직원들에게 공유할 수 있다.

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import requests

def get_binance_news():
    url = "https://api.binance.com/api/v3/ticker/price"
    response = requests.get(url)
    if response.status_code == 200:
        bitcoin_price = response.json().get('BTCUSDT')
        print(f"Binance Bitcoin Price: {bitcoin_price}")
    else:
        print("Failed to fetch Binance Bitcoin price.")

def generate_daily_report():
    # 이 부분에서 일일 레포트를 생성하고 필요한 작업을 수행합니다.
    print("Generating daily report...")

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2022, 1, 1)
}

dag = DAG('binance_bitcoin_report', 
          default_args=default_args,
          description='Daily report on Binance Bitcoin news',
          schedule_interval='@daily'
          )

get_binance_news_task = PythonOperator(
    task_id='get_binance_news',
    python_callable=get_binance_news,
    dag=dag
)

generate_daily_report_task = PythonOperator(
    task_id='generate_daily_report',
    python_callable=generate_daily_report,
    dag=dag
)

get_binance_news_task >> generate_daily_report_task

get_binance_news 함수는 바이낸스에서 비트코인 가격을 가져와 출력하고, generate_daily_report 함수는 일일 레포트를 생성하는 작업을 수행

일일 비트코인 정보 daily batch
가격 정보외에 여러가지 요약 보고서를 받아볼 수 있을 것 같다.

주식정보요약 DAG

: reference by
https://www.inflearn.com/courses/lecture?courseId=331648&type=LECTURE&unitId=173936

Pykrx

PyKrx는 네이버와 KRX에서 주식 관련 정보를 스크래핑 방식으로 가져오는 파이썬 라이브러리이다.
https://github.com/sharebook-kr/pykrx/blob/master/pykrx/stock/stock_api.py
네이버와 KRX 페이지에서 접근하여 주식 관련 데이터를 실시간에 가져온다

pip install pykrx

주식정보 dataframe

시가, 종가 , 고가 , 종가 , 거래량 , 거래대금 등 기본적인 정보를 function통해서 받을 수 있다.
-> 해당 정보를 토대로 gpt에게 보고서 요약 정보를 return

종목의 PER/PBR/배당수익률 조회

지표	의미	계산식	투자 관점에서의 해석
BPS (Book value Per Share)	주당 순자산가치	자기자본 ÷ 발행주식수	기업이 청산될 때 1주당 받을 수 있는 금액
PER (Price Earnings Ratio)	주가수익비율	주가 ÷ EPS	주가가 이익에 비해 고평가/저평가 되었는지 판단
PBR (Price Book Ratio)	주가순자산비율	주가 ÷ BPS	주가가 자산가치에 비해 고평가/저평가 되었는지 판단
EPS (Earnings Per Share)	주당 순이익	당기순이익 ÷ 발행주식수	1주당 벌어들인 이익
DIV (Dividend Yield)	배당수익률	DPS ÷ 주가 × 100	주가 대비 배당금 비율(%)
DPS (Dividend Per Share)	주당 배당금	총배당금 ÷ 발행주식수	1주당 지급되는 현금 배당금