데이터 활용과 투자이야기

sample

from airflow import DAG
import pendulum
from airflow.operators.python import PythonOperator
from airflow.providers.sftp.operators.sftp import SFTPHook, SFTPOperator

sftp_hook = SFTPHook(ssh_conn_id=None, username="SYP", remote_host="xxx.xxx.xxx.xxx", cmd_timeout=None, key_file=None)

with DAG(
        dag_id='hynix_pluto',
        start_date=pendulum.datetime(2024, 7, 1, tz='Asia/Seoul'),
        schedule=None,
        catchup=False
) as dag:
    
    put_file = SFTPOperator(
        task_id="put_file",
        sftp_hook = sftp_hook,
        local_filepath="/opt/airflow/config/10080.py",
        remote_filepath="10080.py",
        operation="get",
        create_intermediate_dirs=True,
        dag=dag
    )

put_file

https://engineer-1.tistory.com/entry/Airflow-SFTP-%EC%A0%91%EC%86%8D-%EC%84%A4%EC%A0%95-SSH-Key-%EC%9D%B8%EC%A6%9D

Extra 옵션 설명 :
no_host_key_check : default 'true' 

- true 일 경우 (airflow 설치한 서버) 로컬 서버의 'known_hosts' 파일에 원격 서버의 정보가 없을 경우 SFTP 접속 시 자동으로 known_hosts에 원격서버의 정보가 작성된다. 
- false 일 경우 로컬 서버의 'known_hosts' 적혀 있지 않은 원격서버의 키 인증 접속은 실패한다. known_hosts 파일의 관리가 필요하다.

단, "no_host_key_check": "true" 일경우 airflow log에 아래 경고 문구가 출력됨으로 업무에 적용 시 참고하길 바란다.

https://github.com/hjkim-sun/airflow/tree/master

▲▲ Python 오퍼레이터에 op_args로 변수 할당하기 ▲▲

dags_python_with_op_args.py

▲▲ Python 오퍼레이터에 op_kwargs로 변수 할당하기 ▲▲

dags_python_with_op_kwargs.py

▲▲ Python Operator에서 Xcom 사용 ▲▲

dags_python_with_xcom_eg1.py

dags_python_with_xcom_eg2.py

▲▲ Bash Operator에서 xcom 사용 ▲▲

dags_bash_with_xcom.py

▲▲ python과 bash operator간에 xcom 변수사용 ▲▲

dags_bash_python_with_xcom.py

▲▲ Bash Operator에서 xcom 사용 ▲▲

dags_bash_with_variable.py

# 20_볼린저밴드_01

import FinanceDataReader as fdr
df = fdr.DataReader('005380', '2015')
df

df = df.loc[:, ['Close']].copy()
df

df['center'] = df['Close'].rolling(window=20).mean()
df

df['up_band'] = df['center'] + 2*df['Close'].rolling(window=20).std()
df

df['low_band'] = df['center'] - 2*df['Close'].rolling(window=20).std()
df

df = df.dropna()
df

# 21_볼린저밴드_02 : 볼린저밴드 지표만들기

book = df[['Close']].copy()
book['trade'] = ''
book

for i in df.index:
    if df.loc[i, 'Close'] > df.loc[i, 'up_band']:
        book.loc[i, 'trade'] = ''
    elif df.loc[i, 'low_band'] > df.loc[i, 'Close']:
        book.loc[i, 'trade'] = 'buy'
    elif df.loc[i, 'up_band'] >= df.loc[i, 'center'] and df.loc[i, 'low_band'] <= df.loc[i,'Close']:
        if book.shift(1).loc[i, 'trade'] == 'buy':
            book.loc[i, 'trade'] = 'buy'
        else:
            book.loc[i, 'trade'] = ''
            
book

23_볼린저밴드_03 : 수익률평가

rtn = 1.0
buy = 0.0
sell = 0.0
book['rtn'] = 1

for i in book.index:
    if book.loc[i, 'trade'] == 'buy' and book.shift(1).loc[i, 'trade'] == '':
        buy = book.loc[i, 'Close']
        print('진입일 : ', i, '매수가격 : ', buy)
    elif book.loc[i, 'trade'] == '' and book.shift(1).loc[i, 'trade'] == 'buy':
        sell = book.loc[i, 'Close']
        rtn = (sell - buy) / buy + 1
        book.loc[i, 'rtn'] = rtn
        print('청산일 : ', i, '매수가격 : ', buy, ' 매도가격 : ', sell, '수익률 : ', round(rtn, 4))
    if book.loc[i, 'trade'] == '':
        buy = 0.0
        sell = 0.0
acc_rtn = 1.0
for i in book.index:
    rtn = book.loc[i, 'rtn']
    acc_rtn = acc_rtn * rtn
    book.loc[i, 'acc_rtn'] = acc_rtn
print('누적수익률 : ', round(acc_rtn, 4))

~~ 중략 ~~

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2000')
df

start_date = '2021-02-04'
end_date = '2022-03-03'
tmp_df = df.loc[start_date:end_date]
tmp_df

tmp_df['daily_rtn'] = tmp_df['Close'].pct_change()
tmp_df

import numpy as np
VOL = np.std(tmp_df['daily_rtn']) * np.sqrt(252.0)
VOL

결과출력 :

np.float64(0.2467837803080943)

SHARP = np.mean(tmp_df['daily_rtn']) / VOL
SHARP

결과출력:

np.float64(0.0033381487604064093)

import FinanceDataReader as fdr
import numpy as np
df = fdr.DataReader('AAPL', '2000')
df

start = '2021-06-13'
end = '2022-02-03'
tmp_df = df.loc[start:end]
tmp_df

tmp_df['Change'] = tmp_df['Close'].pct_change()
tmp_df

VOL = np.std(tmp_df['Change']) * np.sqrt(252.)
VOL

결과출력:

np.float64(0.23523418290202477)

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2000')
df

start_date = '2024-01-22'
end_date = '2025-01-21'
tmp_df = df.loc[start_date:end_date]
tmp_df

historial_max = tmp_df['Close'].cummax()
historial_max
daily_mdd = tmp_df['Close'] / historial_max - 1.0
daily_mdd

mdd = daily_mdd.cummin()
mdd

mdd.plot()

daily_mdd.plot()

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2000')
df

df['daily_rtn'] = df['Close'].pct_change()
df

# cumprod 누적곱, cumsum 누적합
df['st_rtn'] = (df['daily_rtn'] + 1).cumprod()
df

start_date = '2018-10-08'
end_date = '2022-01-12'
tmp_df = df.loc[start_date:end_date, ['st_rtn']] / df.loc[start_date, 'st_rtn']
tmp_df

CAGR = tmp_df.loc['2022-01-12', 'st_rtn'] ** (252./len(tmp_df.index)) -1
CAGR

결과출력 :

np.float64(0.4192544059773151)

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2001-1-20')
df

df['daily_rtn'] = df['Close'].pct_change()
df

# cumprod 누적곱, cumsum 누적합
df['st_rtn'] = (1+df['daily_rtn']).cumprod()
df

start_date = '2001-01-22'
end_date = '2024-05-06'
tmp_df = df.loc[start_date:end_date, ['st_rtn']] / df.loc[start_date, ['st_rtn']]
tmp_df

tmp_df.loc[end_date, 'st_rtn']

import matplotlib.pyplot as plt
tmp_df.plot(figsize=(16,9))

import FinanceDataReader as fdr
df = fdr.DataReader('NAVER:005930', '2024-1-23')
df

day => week, month, year 를 down-sampling 이라고 하고,

주로 사용하는 방식 day => time, minute 를 up-samping 이라고 함

# 1년치 일데이터를 가져온 것을 
# 월(ME:날짜를 어떤날짜로 출력할것인지를 정하는 옵션임
# , MS도 있음 > 이 내용 아래 코딩참조)
df.resample(rule='ME').mean() # mean : 한달치 평균

df.resample(rule='MS').mean()

df.resample(rule='ME').max() # max : 한달치 최대값

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2024-1-20')
df

# 5일 이동평균 구하기
df['MA_5'] = df['Close'].rolling(window=5).mean()
df

# 3일치 최대값 구하기
df['MAX_3'] = df['Close'].rolling(window=3).max()
df.iloc[0:20]