데이터 활용과 투자이야기

import paramiko
import csv

# SSH 접속 정보
host = "remote_server_ip"
username = "user"
password = "password"

# SSH 접속
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname=host, username=username, password=password)

# invoke_shell 을 사용해 명령어 전달
channel = client.invoke_shell()
channel.send("your_command")
time.sleep(30)
output = channel.recv(9999).decode('utf-8')

# output 을 csv 파일로 저장
file_name = "output.csv"
with open(file_name, 'w', newline='') as file:
    writer = csv.writer(file)
    writer.writerow(["output"])
    for line in output:
        writer.writerow([line.strip()])

# SSH 접속 종료
client.close()

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)