Phase 6의 포트폴리오 알고리즘을 이용해서 Splunk 밖에서 무작위의 종목을 골라서 시뮬레이션을 돌린다.
그 후에 그 중에서 가장 좋은 Sharpe 값을 가지 값을 선택해, Splunk에 다시 저장해서 종목을 추천
🌞 핵심🌞
파이썬 용 Splunk SDK를 이용해 파이썬에서 Splunk 데이터 가져오기
HEC를 이용해서 application에서 splunk로 데이터 입력하기
Splunk 매크로 이용하기
🍀 STEP 1 : Splunk Python SDK 설치🍀
cd /opt/splunk/etc/apps/stock/bin
$ pip3 install splunk-sdk
OR
$ pip install splunk-sdk
OR
$ apt get install splunk-sdk
설치를 어디에다가 꼭 하라는 말은 없었지만
나는 stock 앱 밑에 bin 밑에 이미 splunk_sdk가 있어서 생략해줬다.
🍀STEP 2 : Splunk에서 데이터를 읽을 수 있는 부분 만들기🍀
어디에다가 꼭 하라는 말이 없어서 /opt/splunk/etc/apps/stock/bin 밑에 파일을 생성해줬다.
$ vi splunk_data_reader.py
# Splunk 에서 데이터를 불러오기위한 라이브러리 import
import splunklib.results as results
import splunklib.client as client
import io, os, sys, types, datetime, time
import pandas as pd
"""
스플렁크에 SPL 을 수행해서 데이터를 가져온다.
@host : 스플렁크 설치 호스트
@port: 스플렁크 REST API 접속 포트 (8089)
@username: 스플렁크 접속 계정
@password: 스플렁크 접속 패스워드
"""
class SplunkDataReader():
def __init__(self, host, port, username, password):
self._host = host
self._port = port
self._username = username
self._password = password
self._service = None
"""
스플렁크에 접속한다.
"""
def connect(self):
self._service = client.connect(host=self._host,
port = self._port,
username = self._username,
password = self._password);
"""
스플렁크에 SPL 을 수행해서 데이터를 읽어서 결과를 리턴한다.
@searchquery_normal: 일반적인 SPL 문
"""
def execute_query(self, searchquery_normal,
kwargs_normalsearch={"exec_mode":"normal"},
kwargs_options={"output_mode":"csv", "count":100000}):
# 스플렁크에 SPL 을 수행하면 해당 job 이 생성된다. 이 Job을 통해 async 하게 데이터를 가져와야 한다.
job = self._service.jobs.create(searchquery_normal, **kwargs_normalsearch)
# 모든 작업이 끝날 때까지 대기하기 위해 Loop 수행 (@TODO 여러 쿼리를 수행해야 하는 경우 Thread 처리해야 함)
while True:
# 작업이 실행 준비 되었는지 체크하고 아니면 계속 모니터링
while not job.is_ready():
pass
# 현재 작업이 수행 중에 있고, 작업의 상태를 모니터링 한다.
stats = {"isDone":job["isDone"], "doneProgress":float(job["doneProgress"]) *100,
"scanCount":int(job["scanCount"]), "eventCount":int(job["eventCount"]),
"resultCount":int(job["resultCount"])}
status = ("\r%(doneProgress)03.1f%% %(scanCount)d scanned"
"%(eventCount)d matched %(resultCount)d results") % stats
# 작업 상태를 사용자를 위해 출력
sys.stdout.write(status)
sys.stdout.flush()
# 작업 상태가 완료가 되었다면 루프를 빠져나감
if stats["isDone"] == "1":
sys.stdout.write("\nDone!")
break;
time.sleep(0.5)
# 작업 결과를 받아오고, 데이터프레임에 저장
csv_results = job.results(**kwargs_options).read()
df = pd.read_csv(io.BytesIO(csv_results), encoding='utf8', sep=',')
# 작업을 제거
job.cancel()
# 결과 리턴
return df
간단하다는데 전혀 간단해보이지 않아서 당황스럽다.
그런데 실제 product에서 사용하기 위해서는 예외 처리, 쓰레드 처리 등 더 많은 작업이 필요하다고 한다.
🍀STEP 3 : API 연결 설정 (feat. server.conf)🍀
$SPLUNK_HOME/etc/system/local/에 server.conf 파일을 수정해야함.
일단 아래 위치에 server.conf 파일이 있는지 부터 확인해보자.
아, 그 전에 환경변수부터 설정을 좀 해보려고 한다.
# 실행할 계정의 .bashrc를 수정한다.
vi ~/.bashrc # 로 문서를 연 후
# 끝 부분에 다음과 같은 내용을 추가해 넣는다.
export SPLUNK_HOME=/opt/splunk # Splunk 설치 경로
export PATH=${PATH}:${SPLUNK_HOME}/bin
# 내용을 추가해 넣은 다음 source ~/.barshrc를 해준다.
# 잘 설정되었는지 확인하기
echo $SPLUNK_HOME
cd $SPLUNK_HOME
성공
cd $SPLUNK_HOME/etc/system/local
원래 있었구만!
이제 진짜 Splunk와 API를 연결하기 위한 설정을 한 번 해보자.
$ vi $SPLUNK_HOME/etc/system/local/server.conf
[general]
...
allowRemoteLogin = always
오타에 주의하며 잘 작성을 해준다.
🍀STEP 4 : Montecarlo Simulation 코드 입력🍀
이 코드는 Phase 6에서 이미 작성한 것에서 조금 수정을 했다.
마찬가지로 $SPLUNK_HOME/etc/apps/stock/bin/에 넣었다.
$ vi monte_sim.py
import pandas as pd
import numpy as np
class MonteCarloSim():
def __init__(self, numiter=5000):
self.counter = numiter
def fit(self, df):
""" Compute the Monte Carlo Simulator """
# df contains all the search results, including hidden fields
# but the requested are saved as self.feature_variables
df.sort_values(by=['date'])
df.set_index('date', inplace=True)
codes = df.columns
days = df.shape[0] / len(codes)
daily_ret = df.pct_change()
annual_ret = daily_ret.mean() * days
daily_cov = daily_ret.cov()
annual_cov = daily_cov * days
port_ret = []
port_risk = []
port_weights = []
port_sharpe = []
for _ in range(self.counter):
weights = np.random.random(len(codes))
weights /= np.sum(weights)
returns = np.dot(weights, annual_ret)
risk = np.sqrt(np.dot(weights.T, np.dot(annual_cov, weights)))
port_ret.append(returns)
port_risk.append(risk)
port_weights.append(weights)
port_sharpe.append(returns/risk)
portfolio = { 'Sharpe': port_sharpe, 'Returns' : port_ret, 'Risk': port_risk }
for i, s in enumerate(codes):
portfolio[s] = [weight[i] for weight in port_weights]
output_df = pd.DataFrame(portfolio)
output_df = output_df[['Sharpe', 'Returns', 'Risk'] + [s for s in codes]]
return output_df
🍀STEP 5 : 실제로 작업을 수행하는 코드 입력🍀
main 함수가 포함되어 있다.
소스는 길지만 naive한 소스라서 이해하기는 어렵지 않다고 한다.
마찬가지로 $SPLUNK_HOME/etc/apps/stock/bin/에 넣었다.
아래 코드의 To do 부분을 직접 작성해야한다.
$ vi run_portpolio.py
# 방금 만들 파이썬 파일들 import
from splunk_data_reader import SplunkDataReader
from monte_sim import MonteCarloSim
import pandas as pd
import numpy as np
import argparse
from datetime import datetime, timedelta
import json
import os
import sys
from random import seed
from random import randinte
import time
import requests
import urllib3
# kospi_200. csv 파일을 읽기 위한 경로 설정
CONF_PATH=os.path.dirname(os.path.abspath(__file__))
LOG_PATH=CONF_PATH + '/stock'
# kospi_200.csv 에서 주식 종목 리스트를 읽어온다.
def getCodeList():
df = pd.read_csv(CONF_PATH + "/kospi_200.csv", dtype=str)
return df.code.values
seed(datetime.now()) # 랜덤 시드 초기화
"""
kospi 200 목록에서 종목을 선택한다.
@codebook[] : 전체 종목 리스트
@codes[]: 필수적으로 포함시키고 싶은 종목 목록
@ num: 총 선택할 종목 수
"""
def genRandomCode(codebook, codes, num):
random_ranges = len(codebook)
# num=5 이고 필수적으로 포함시켜향 종목의 수가 2 이면 3개의 종목이 랜덤하게 선택된다.
itera = num - len(codes)
ran_code = [ code for code in codes]
for _ in range(itera):
while True:
# 랜덤하게 종목 코드 하나 선택
code = codebook[randint(0, random_ranges-1)] + ".KS"
# 이미 포함 된 코드이면 pass
if code in ran_code:
pass
# 새로운 코드면 추가
else:
ran_code.append(code)
break
return ran_code
"""
스플렁크에서 필요한 데이터를 읽어온다.
@reader: 스플렁크 접속 객체
@codes[]: 주식 종목 리스트
@days: 가져올 데이터의 날 수
"""
def readData(reader, codes, days=180):
#스플렁크 SPL 문
splunk_query="""
search index=kospi {codes} earliest={days}
| rename Date as date
| chart latest(Close) as value by date, code
"""
code_str = " OR ".join(codes)
days_str = "-" + str(days) + "d@"
# SPL을 수행해서 데이터를 가져온다.
df = reader.execute_query(splunk_query.format(codes=code_str, days=days_str))
return df
"""
HEC 를 통해 스플렁크에 결과 데이터를 저장한다.
@host: 스플렁크 접속 서버
@token: 스플렁크에서 발급한 접속 토근
"""
authToken = "XXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX" # @TODO 나의 토큰으로 대체
splunkhost = "localhost" # @TODO 나의 서버 주소로 대체
def splunkHec(host, token, data):
url="https://" + host + ":8088/services/collector/event"
authHeader = {'Authorization': 'Splunk ' + token}
payload = {}
payload.update({"index":"monte"}) # "monte" 라는 인덱스에 저장
payload.update({"sourcetype":"_json"}) # "_json" 소스타입으로 저장
payload.update({"source":"monte_sim.py"}) # "monte_sim.py" 소스로 저장
payload.update({"event": data}) # "저장할 데이터 메시지"
r = requests.post(url, headers=authHeader, json=payload, verify=False) # 스플렁크로 데이터 전송
"""
반복해서 임의의 코드 종목을 선택하여 최상의 종목 조합을 찾음
@ reader: 스플렁크 접속 객체
@ codebook: 주식 종목 리스트
@ required_codes: 반드시 포트폴리오에 포함해야할 코드 목록
@ num: 포트폴리오를 구성할 주식 종목의 수
@ repeat: 반복 횟수
"""
def findBestPortpolio(reader, codebook, required_codes, num, repeat=10000):
# 몬테카를로 시뮬레이터 객체 생성
sim = MonteCarloSim()
# repeat 만큼 반복해서 주식 종목 선택 수행
for idx in range(repeat):
# 랜덤한 주식 종목을 가져온다.
c = genRandomCode(codebook, required_codes, num)
# 선택된 주식 종목에 대해서 스플렁크에서 해당 데이터를 가져온다.
df = readData(reader, c, days)
# 몬테카를로 시뮬레이터를 수행한다.
monte_df = sim.fit(df)
# Sharpe 값이 가장 좋은 항목을 선택한다.
monte_max = monte_df.loc[monte_df['Sharpe'].idxmax()]
# 선택된 데이터에 대해서 스플렁크에 전달할 데이터 모양을 만든다.
dic = {}
codes = []
rates = []
for key in monte_max.keys():
if key not in ['Sharpe', 'Returns', 'Risk']:
codes.append(key)
rates.append(monte_max[key])
else:
dic[key] = monte_max[key]
dic['idx'] = idx
dic["code"] = codes
dic["rate"] = rates
dic["date"] = datetime.now().strftime("%Y-%m-%d")
data = json.dumps(dic)
# 스플렁크로 해당 데이터를 보낸다.
splunkHec(splunkhost, authToken, data)
# 인자로 전달되는 코드의 유효성 체크
def verifyCode(required):
codes = required.split(",")
codes = [ v + ".KS" if len(v)==6 else v for v in codes]
return codes;
if __name__ == "__main__":
# 전달받는 파라미터 설정
parser = argparse.ArgumentParser()
parser.add_argument('--days', help='days help')
parser.add_argument('--num', help='num help')
parser.add_argument('--required', help='required help')
parser.add_argument('--repeat', help='repeat help')
args = parser.parse_args()
if args.days:
days = int(args.days)
else:
days = 180
if args.num:
num = int(args.num)
else:
num = 5
if args.repeat:
repeat = int(args.repeat)
else:
repeat = 1000
codes =[]
if args.required:
codes = verifyCode(args.required);
# 주식 종목 리스트 읽기
codebook = getCodeList()
# 스플렁크 접속 객체 생성 @TODO 나의 접속 정보로 수정
reader = SplunkDataReader("localhost", 8089, "my_account", "my_password")
# 스플렁크에 접속
reader.connect()
# 포트 폴리오 검색 수행
findBestPortpolio(reader, codebook, codes, num, repeat)
여기까지 했다면 splunk를 restart 해준다.(.conf를 수정한 이력이 있다면 splunk 재시작해야한다.)
줄번호가 보이게 하기 위해서 ESC 누른 후 :set number 또는 :set nu 을 입력한다.
가독성이 증가했다.
randinte가 아니라 randint다 변경해주자
root@splunkenterprise:/opt/splunk/etc/apps/stock/bin# python3 run_portpolio.py --days 350 --num 5 --repeat 1000
/opt/splunk/etc/apps/stock/bin/run_portpolio.py:26:
DeprecationWarning: Seeding based on hashing is deprecated
since Python 3.9 and will be removed in a subsequent version.
The only supported seed types are: None, int, float, str, bytes, and bytearray.
seed(datetime.now()) # 랜덤 시드 초기화
Traceback (most recent call last):
File "/opt/splunk/etc/apps/stock/bin/run_portpolio.py", line 168, in <module>
codebook = getCodeList()
File "/opt/splunk/etc/apps/stock/bin/run_portpolio.py", line 23, in getCodeList
df = pd.read_csv(CONF_PATH + "/kospi_200.csv", dtype=str)
File "/usr/local/lib/python3.10/dist-packages/pandas/io/parsers/readers.py", line 912, in read_csv
return _read(filepath_or_buffer, kwds)
File "/usr/local/lib/python3.10/dist-packages/pandas/io/parsers/readers.py", line 577, in _read
parser = TextFileReader(filepath_or_buffer, **kwds)
File "/usr/local/lib/python3.10/dist-packages/pandas/io/parsers/readers.py", line 1407, in __init__
self._engine = self._make_engine(f, self.engine)
File "/usr/local/lib/python3.10/dist-packages/pandas/io/parsers/readers.py", line 1661, in _make_engine
self.handles = get_handle(
File "/usr/local/lib/python3.10/dist-packages/pandas/io/common.py", line 859, in get_handle
handle = open(
FileNotFoundError: [Errno 2] No such file or directory: '/opt/splunk/etc/apps/stock/bin/kospi_200.csv'
이런 에러 메시지가 나왔다.
첫번째는, 일단 버전 문제가 확실히 있는 것 같다.
DeprecationWarning: Seeding based on hashing is deprecated since Python 3.9 and will be removed in a subsequent version. The only supported seed types are: None, int, float, str, bytes, and bytearray.
해싱 기반 시딩은 파이썬 3.9 버전 부터 사용되지 않고 제거되었다고 한다.
내껀 3.10.6 버전임 (python3 -version으로 Linux환경에서 간단하게 확인 가능)
지원되는 유일한 시드 유형은 None, 정수, 소수, 문자열, bytes, bytearray라고 한다.
두번째는, kospi_200.csv 파일이 /opt/splunk/etc/apps/stock/bin/kospi_200.csv 에 없는 것이 문제다.
🔔 첫번째 오류 해결하기
보니 입력되어 있는 형태가 timedate() 형태라서 문제가 발생한 것 같다.
근데 수정 없이 해결되었음... 할 수 없다.. 정말.. 자동으로 string으로 전환된걸까?
