시세데이타(배치) 처리 개발

거래소로 부터 수신한 데이타를 처리하는 프로세스를 개발한다.

처리프로세스

데이타	처리프로세스	기능
배치/기타	sise_job_grp1	데이타를 가공해서 postgres,redis에 저장
체결	sise_job_a3	postgres에 적재 / 마스타는 redis에 업데이트
호가	sise_job_b6	redis에 업데이트
장운영	sise_job_grp4	redis에 업데이트

배치(마스타)/기타 데이타를 처리하는 sise_job_grp1를 개발

배치/기타 데이타중 가장 기본이 되고 가장 먼저 수신되는 증권종목정보(A0)를 처리한다.

처음으로 log를 처리하기 위한 작업이다. k8s에서는 콘솔출력(stdout, stderr)을 사용할 수 있어서 로그파일을 만들어서 작업할 필요가 많이 없어졌다. initApLogSt()기능은 파라메타에 따라 여러 기능을 제공한다.

// USE_LOG = default:local-FILE // 0:local-FILE, 1:IPC-FILE, 2:KAFKA-MongoDB, 9:stderr if ((_gLog = initApLogSt((char *)hostname, log_mode, LOG_FILE_DEFAULT, NULL, NULL, NULL, (char *)hostname, (char *)conf_method)) == NULL) { fprintf(stderr, "[%s][%s][%d]:initApLogSt error\n", __FILE__, __FUNCTION__, __LINE__); return -1; }

로그파일로 출력, 메세지큐나 kafka로 보내는 방법, 콘솔로 출력하는 방법이 있는데 여기서 사용하는 방법은 log_mode = 9 콘솔출력으로 설정한다.

다음으로는 UDP수신 프로세스에서 데이타를 받은 방법을 설정한다. MSG_BROKER_TYPE=zmq로 ZeroMQ로 수신 한다. ZeroMQ로 사용하는 방법은 테스트를 해본 결과 ZeroMQ가 가장 효율적이라고 판단해서다.

// zmq 설정 char *zmqServer = getenv("ZMQ_SERVER"); _gZmq = initApZmqSt(_gLog->pLog, ZMQ_SERVER, zmqServer, ZMQ_ROUTER, HWM); if (_gZmq == NULL) { fprintf(stderr, "initApZmq error: [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; }

DB는 db_type에 따라 MySQL, PostgreSQL 둘중 하나로 설정을 한다. initApSqlSt()에서 db 연결을 한다.

_gSql = initApSqlSt(_gLog->pLog, db_type, db_host, db_port, db_user, db_password, db_name); if (_gSql == NULL) { fprintf(stderr, "[%s][%s][%d]:initApSqlSt error\n", __FILE__, __FUNCTION__, __LINE__); return -1; }

Redis 연결을 한다. 메인과 스레드에서 사용하기 때문에 메인용, 스레드용 2개의 연결을 생성한다.

// 스레드 전용 if ((_gRedisThread = initApRedisSt(_gLog->pLog, redis_host, redis_port, NULL)) == NULL) { fprintf(stderr, "initApRedisSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } // 메인 전용 if ((_gRedisMain = initApRedisSt(_gLog->pLog, redis_host, redis_port, NULL)) == NULL) { fprintf(stderr, "initApRedisSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; }

메인에서 데이타를 수신해서 데이타 가공후에 db처리하는 스레드와 redis처리하는 스레드에 데이타를 송신하고 스레드에서 수신하는 채널을 생성한다.

// db처리하는 스레드와의 통신 if ((_gPgChannel = initChannelSt(QUEUE_SIZE, BUFFER_SIZE)) == NULL) { fprintf(stderr, "PgChannel initChannelSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } // redis처리하는 스레드와의 통신 if ((_gRedisChannel = initChannelSt(QUEUE_SIZE, BUFFER_SIZE)) == NULL) { fprintf(stderr, "RedisChannel initChannelSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; }

타파스의 시세처리 프로세스의 구성은

함수	기능
init()	log,메세지브로커,db,redis 등을 생성
dispose()	init()에서 생성된 자원을 해제
requestProc()	시세데이타를 수신 받아서 처리하는 함수
main()	메인함수

크게 4개의 함수로 이루어져 있다.

위에서 살펴본 초기함수들을 init()안에 기술한다. 전체 소스는 아래와 같다.

// 환경파일에서 로그, 메시지큐, 데이터베이스, 등 설정 // k8s에서는 configmap.yml에서 설정 static int _init(char *pname) { // log const char *log_topic = getenv("LOG_TOPIC"); // logs const char *hostname = getenv("HOSTNAME"); // 별도의 conf파일을 사용하는지 여부 const char *conf_method = getenv("CONF_METHOD"); // USE_LOG = default:local-FILE // 0:local-FILE, 1:IPC-FILE, 2:KAFKA-MongoDB, 9:stderr // k8s에서 로그를 전송해서 모니터링하는 경우가 많아서 stderr를 사용 const int log_mode = atoi(getenv(LOG_MODE)); if ((_gLog = initApLogSt((char *)hostname, log_mode, LOG_FILE_DEFAULT, NULL, NULL, NULL, (char *)hostname, (char *)conf_method)) == NULL) { fprintf(stderr, "[%s][%s][%d]:initApLogSt error\n", __FILE__, __FUNCTION__, __LINE__); return -1; } _gLog->pLog->putLog(_gLog->pLog, (char *)"start init : %s\n", hostname); // 메세지브로커 처리하기 위한 설정 char *msgBrokerType = getenv("MSG_BROKER_TYPE"); if (_gMsgBrokerType == NULL) { _gLog->pLog->putLog(_gLog->pLog, (char *)"message broker is null error!\n"); return -1; } if (strcmp(msgBrokerType, "zmq") != 0) { // 메세지브로커로 kafka/rabbitmq를 사용하는설정 MSG_BROKER_ST msgBrokerSt; memset(&msgBrokerSt, 0, sizeof(MSG_BROKER_ST)); char *msgBrokerHost = getenv("MSG_BROKER_HOST"); int msgBrokerPort = atoi(getenv("MSG_BROKER_PORT")); strncpy(msgBrokerSt.hostname, msgBrokerHost, sizeof(msgBrokerSt.hostname) - 1); msgBrokerSt.port = msgBrokerPort; if (strcmp(msgBrokerType, "kafka") == 0) { msgBrokerSt.brokerType = KAFKA; msgBrokerSt.type = RD_KAFKA_PRODUCER; msgBrokerSt.dr_msg_cb = NULL; } else if (strcmp(msgBrokerType, "rabbitmq") == 0) { msgBrokerSt.brokerType = RABBITMQ; char *msgBrokerUser = getenv("MSG_BROKER_USER"); char *msgBrokerPass = getenv("MSG_BROKER_PASS"); msgBrokerSt.channel = atoi(getenv("RABBITMQ_CHANNEL")); strncpy(msgBrokerSt.user, msgBrokerUser, sizeof(msgBrokerSt.user) - 1); strncpy(msgBrokerSt.pass, msgBrokerPass, sizeof(msgBrokerSt.pass) - 1); } else { fprintf(stderr, "message broker[%s] type error!\n", msgBrokerType); return -1; } fprintf(stderr, "initApMsgBroker send.....\n"); _gMsgBrokerSend = initApMsgBroker(_gLog->pLog, &msgBrokerSt); if (_gMsgBrokerSend == NULL) { fprintf(stderr, "initApMsgBroker send error: [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } if (strcmp(msgBrokerType, "kafka") == 0) { msgBrokerSt.type = RD_KAFKA_CONSUMER; // groupId char *msgGroupId = getenv("KAFKA_GROUPID"); strncpy(msgBrokerSt.groupId, msgGroupId, sizeof(msgBrokerSt.groupId) - 1); // offset_reset char *msgOffset = getenv("KAFKA_OFFSET"); strncpy(msgBrokerSt.offset_reset, msgOffset, sizeof(msgBrokerSt.offset_reset) - 1); } fprintf(stderr, "initApMsgBroker recv.....\n"); _gMsgBrokerRecv = initApMsgBroker(_gLog->pLog, &msgBrokerSt); if (_gMsgBrokerRecv == NULL) { fprintf(stderr, "initApMsgBroker recv error: [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } } else { // zmq 설정 char *zmqServer = getenv("ZMQ_SERVER"); _gZmq = initApZmqSt(_gLog->pLog, ZMQ_SERVER, zmqServer, ZMQ_ROUTER, HWM); if (_gZmq == NULL) { fprintf(stderr, "initApZmq error: [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } } // db 설정 : mysql, postgres char *db_type = getenv("DB_TYPE"); // mysql, postgres char *db_host = getenv("DB_HOST"); char *db_user = getenv("DB_USER"); // char *db_password = getenv("MYSQL_ROOT_PASSWORD"); char *db_password = getenv("DB_PASS"); char *db_name = getenv("DB_NAME"); int db_port = atoi(getenv("DB_PORT")); _gSql = initApSqlSt(_gLog->pLog, db_type, db_host, db_port, db_user, db_password, db_name); if (_gSql == NULL) { fprintf(stderr, "[%s][%s][%d]:initApSqlSt error\n", __FILE__, __FUNCTION__, __LINE__); return -1; } // redis 설정 char *redis_host = getenv("REDIS_HOST"); int redis_port = atoi(getenv("REDIS_PORT")); // redis update전용 if ((_gRedis = initApRedisSt(_gLog->pLog, redis_host, redis_port, NULL)) == NULL) { fprintf(stderr, "initApRedisSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } // redis read전용 if ((_gRedisSeq = initApRedisSt(_gLog->pLog, redis_host, redis_port, NULL)) == NULL) { fprintf(stderr, "initApRedisSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } // db처리하는 스레드와의 통신 if ((_gPgChannel = initChannelSt(QUEUE_SIZE, BUFFER_SIZE)) == NULL) { fprintf(stderr, "PgChannel initChannelSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } // redis처리하는 스레드와의 통신 if ((_gRedisChannel = initChannelSt(QUEUE_SIZE, BUFFER_SIZE)) == NULL) { fprintf(stderr, "RedisChannel initChannelSt : [%s][%s][%d]\n", __FILE__, __FUNCTION__, __LINE__); return -1; } // pflib 사용하기위한 처리 if ((_gPfLib = initPfLibSt()) == NULL) { return -1; } return 0; }

_init()내 주요한 함수는 아래와 같다.

함수	Lib	기능
initApLogSt	aplib	로그를 처리하기 위한 함수
initApMsgBroker	aplib	메세지 브로커를 생성한다. 지원하는 메세지 브로커는 kafka, rabbitmq이다.
initApZmqSt	aplib	ZeroMQ를 사용하기 위한 함수
initApSqlSt	aplib	사용할 db를 연결한다. 지원하는 db는 mysql, postgresql이다.
initApRedisSt	aplib	사용할 redis를 연결한다.
initChannelSt	pflib	스레드와의 통신을 위해서 사용
이함수는 별도로 자세히 기술하기로 한다.

모든 프로그램에서 _init()함수 사용후 반드시 _dispose()함수를 반드시 호출해야 한다. aplib/pflib의 라이브러리, 타파스 모든 프로세스는 init()으로 생성해서 dispose()로 해제하게 설계 되었다.

/* * 사용한 메모리/자원을 해제함 */ void _dispose() { if (_gPgChannel) { _gPgChannel->dispose(_gPgChannel); _gPgChannel = NULL; } if (_gRedisChannel) { _gRedisChannel->dispose(_gRedisChannel); _gRedisChannel = NULL; } if (_gZmq) { _gZmq->dispose(_gZmq); _gZmq = NULL; } if (_gRedis) { _gRedis->dispose(_gRedis); _gRedis = NULL; } if (_gRedisSeq) { _gRedisSeq->dispose(_gRedisSeq); _gRedisSeq = NULL; } if (_gMsgBrokerRecv) { _gMsgBrokerRecv->dispose(_gMsgBrokerRecv); _gMsgBrokerRecv = NULL; } if (_gMsgBrokerSend) { _gMsgBrokerSend->dispose(_gMsgBrokerSend); _gMsgBrokerSend = NULL; } if (_gLog) { _gLog->dispose(_gLog); _gLog = NULL; } if (_gPfLib) { _gPfLib->dispose(); } if (_gSql) { _gSql->dispose(_gSql); } }

_init()함수내에서 메모리를 할당했거나, 메세지브로커, db에 연결을 했다면 _dispose()함수에서는 메모리를 해제, 연결을 끊는 처리를 한다.

지금까지 개발을 할때 별도의 환경파일을 만들어서 설정을 하곤 했다. 그런데 k8s환경에서는 그다지 효율적이지 못할것으로 판단했다.

그래서 결정한 방법이 k8s의 configmap을 사용한다. 아래는 k8s에서 sise_job_grp1에서 사용하는 configmap이다. _init()함수내에서 getenv()로 참조하면 된다.

apiVersion: v1 kind: ConfigMap metadata: name: sise-job-grp1-config namespace: tapas data: WORKHOME : "/opt" # message broker MSG_BROKER_TYPE : "zmq" #msgq, kafka, rabbitmq ZMQ_SERVER : "tcp://*:5501" #msgq, kafka, rabbitmq # MSG_BROKER_HOST : "msg-broker" MSG_BROKER_HOST : "host.docker.internal" MSG_BROKER_PORT : "5672" #kafka:9092, rabbitmq:5672(외부), 30672(내부) MSG_BROKER_USER : "guest" #rabbitmq user MSG_BROKER_PASS : "guest" #rabbitmq password # # rabbitmq의 exchange/routing key EXCHANGE : "sise.grp1" ROUTING_KEY : "sise.1" # LOG_TOPIC : "sise-job-grp1.logs" # conf -> env CONF_METHOD : "ENV" LOG_MODE : "9" #0:local-FILE, 1:IPC(message queue):FILE, 2:KAFKA:MongoDB, 9:stderr # topic TOPIC_ETC : "sise.etc" #그외 토픽 TOPIC_ETC_RETRY : "sise.etc.retry" #재전송 토픽(오류대응) # for kafka GROUPID : "sise-etc-grp" # db정보 DB_TYPE : "postgres" #db type정보 DB_HOST : "host.docker.internal" #db host정보 DB_NAME : "postgres" #db의 사용 db DB_USER : "postgres" #db의 유저 DB_PASS : "dayu1202" #db의 비밀번호 DB_PORT : "5432" #db nodeport # redis-cluster정보 REDIS_CNT : "1" #redis-cluster 갯수 #REDIS_HOST : "host.docker.internal" #redis-cluster host정보 REDIS_HOST : "192.168.1.59" #redis-cluster host정보 REDIS_PORT : "7617" #redis-cluster nodeport REDIS_HOST2 : "10.104.43.26" #redis-cluster host정보 REDIS_PORT2 : "6379" #redis nodeport

_init()함수에서 사용하는 환경변수는 configmap에 정의하면 된다.

UDP데이타 수신(클라이언트) -----> 데이타처리(서버:sise_job_grp1)에서 통신은 ZeroMQ를 사용한다.

MSG_BROKER_TYPE : "zmq"으로 ZeroMQ로 사용하고 ZMQ_SERVER로 설정했다. 메세지브로커로 사용하고 싶으면 MSG_BROKER_TYPE : "kafka", "rabbitmq"로 설정하면 된다.

ZeroMQ를 사용하므로 UDP데이타를 수신하는 쪽에서 먼저 sise_job_grp1에 연결을 하고 UDP 데이타를 받아서 sise_job_grp1에 송신한다.

그래서 UDP데이타 수신 프로세스 보다 sise_job_grp1이 먼저 실행되어 수신프로세스에서 sise_job_grp1으로 연결을 해야 한다.

이제 _init()함수에서 사용하는 변수와 main()함수를 보자

static CHANNEL_ST *_gPgChannel; static CHANNEL_ST *_gRedisChannel; static PFLIB_ST *_gPfLib; static APLOG_ST *_gLog; // Log // static APMSG_BROKER_ST *_gMsgBrokerRecv; static APMSG_BROKER_ST *_gMsgBrokerSend; static APSQL_ST *_gSql; static APREDIS_ST *_gRedis; static APREDIS_ST *_gRedisSeq; static APZMQ_ST *_gZmq; static int _gIsExit = 0; #define HWM 10000 // 큐 크기 // pg, redis 스레드처리 #define BUFFER_SIZE 1024 * 2 #define QUEUE_SIZE 100000 // 큐 크기 // // Name Utf Size : 종목명(40) * 2 #define NAME_UTF8_SIZE 40 * 2 int main(int argc, char **argv) { pthread_t pg_thread_t, redis_thread_t; _setSignal(); setpgrp(); if (_init(argv[0]) < 0) { _dispose(); return -1; } // db처리하는 스레드와의 통신 pthread_create(&pg_thread_t, NULL, pg_thread, _gPgChannel); // redis처리하는 스레드와의 통신 pthread_create(&redis_thread_t, NULL, redis_thread, _gRedisChannel); // 시세데이타를 처리할 유형 if (_gMsgBrokerRecv) { // exchange, routing key(topic) char *exchange = getenv("EXCHANGE"); char *routing_key = getenv("ROUTING_KEY"); // 처리할 topic char topics[1][TOPIC_LEN]; strcpy(topics[0], routing_key); // 메세지브로커로 부터 데이타를 받으면 requestProc()를 호출한다. _gMsgBrokerRecv->recv(_gMsgBrokerRecv, topics, 1, requestProc, _gMsgBrokerRecv->channel, exchange, "direct"); } else { if (_gZmq) { // ZMQ로 부터 데이타를 받으면 requestProc()를 호출한다. _gZmq->recv(_gZmq, requestProc); } } pthread_join(pg_thread_t, NULL); pthread_join(redis_thread_t, NULL); while (!_gIsExit) { } // 자원을 해제하는 함수 _dispose(); return 0; }

메인함수에서 스레드와 통신하기 위해 생성한 채널(_gPgChannel, _gRedisChannel)을 스레드 생성시 파라메타로 넘겨준다.

// db처리하는 스레드와의 통신 pthread_create(&pg_thread_t, NULL, pg_thread, _gPgChannel); // redis처리하는 스레드와의 통신 pthread_create(&redis_thread_t, NULL, redis_thread, _gRedisChannel);

sise_job_grp1에서 사용하는 모든 자원은 static으로 외부변수로 선언했다. main()에서 _init()함수가 성공적이라면 db처리 스레드와 redis처리 스레드를 생성하고 메세지브로커나 ZeroMQ로 부터 데이타가 들어올때 까지 대기한다.

데이타가 들어오면 requestProc()함수가 호출된다. 메세지브로커 사용시 수신 받을 함수(requestProc)를 넘겨준다.

// 메세지브로커로 부터 데이타를 받으면 requestProc()를 호출한다. _gMsgBrokerRecv->recv(_gMsgBrokerRecv, topics, 1, requestProc, _gMsgBrokerRecv->channel, exchange, "direct");

ZeroMQ 사용시 수신 받을 함수(requestProc)를 넘겨준다.

// ZMQ로 부터 데이타를 받으면 requestProc()를 호출한다. _gZmq->recv(_gZmq, requestProc);

데이타를 처리하는 모든 로직은 requestProc()함수에서 처리하면 된다.

int requestProc(void *msgBuf, size_t msgLen, int channel) { int ret; char *pRcvData = (char *)msgBuf; // recv_cnt++; //_gLog->putLog(_gLog, __FILE__, __FUNCTION__, __LINE__, LOG_TYPE_INFO, 255, //"[%d]:[%20.20s]\n", recv_cnt, pRcvData); // return 0; // 데이타가공처리 if (!memcmp(pRcvData, A0, DATA_DIV_LEN)) { char data[BUFFER_SIZE]; size_t len; char utf8KorName[NAME_UTF8_SIZE]; // UTF8->1.5배 char utf8EngName[NAME_UTF8_SIZE]; // UTF8->1.5배 KOSCOM_STOCK_A0 *pA0 = (KOSCOM_STOCK_A0 *)pRcvData; recv_cnt++; memset(utf8KorName, 0x00, sizeof(utf8KorName)); memset(utf8EngName, 0x00, sizeof(utf8EngName)); ret = make_sise_a0(_gLog->pLog, pRcvData, utf8KorName, utf8EngName, NAME_UTF8_SIZE - 1, _gPfLib); if (ret == 1) { // 종목배치를 전부 수신한 경우의 처리 (isin=999999999999) // redis 일련번호초기화 _gLog->putLog(_gLog, __FILE__, __FUNCTION__, __LINE__, LOG_TYPE_INFO, 255, "recv_cnt[%d]:[%17.17s]\n", recv_cnt, pRcvData); if (_gRedisMain->command(_gRedisMain, "SET %s %d", A3_SEQ, 0) < 0) { _gLog->pLog->putLog(_gLog->pLog, (char *)"set stock_a3_sequence error...\n"); } // 영업일설정 if (_gRedisMain->command(_gRedisMain, "SET %s %s", BIZ_DAY, _gBizDay) < 0) { _gLog->pLog->putLog(_gLog->pLog, (char *)"set biz_day error...\n"); } return 0; } else if (ret < 0) { return -1; } // 영업일설정 if (strlen(_gBizDay) == 0) { memcpy(_gBizDay, pA0->biz_day, sizeof(pA0->biz_day)); } // 스레드로 보낼 데이타 생성 : A0전문 + utf8KorName + utf8EngName // msgLen -1 : newline제외 // fprintf(stderr, "msgBuf[%ld:%d][%s]\n", msgLen, (int)SZ_KOSCOM_STOCK_A0, // (char *)msgBuf); memset(data, 0x00, sizeof(data)); memcpy(data, msgBuf, msgLen - 1); len = SZ_KOSCOM_STOCK_A0; // KorName memcpy(data + len, utf8KorName, NAME_UTF8_SIZE); len += NAME_UTF8_SIZE; // EngName memcpy(data + len, utf8EngName, NAME_UTF8_SIZE); len += NAME_UTF8_SIZE; // redis처리 _gRedisChannel->send(_gRedisChannel, data, len); // postgres처리 _gPgChannel->send(_gPgChannel, data, len); return 0; } else if (!memcmp(pRcvData, B5, DATA_DIV_LEN)) { // 현재등락 return 0; } else if (!memcmp(pRcvData, I5, DATA_DIV_LEN)) { // 결산일정보 return 0; } else if (!memcmp(pRcvData, I6, DATA_DIV_LEN)) { // 종목이벤트정보 return 0; } else if (!memcmp(pRcvData, I7, DATA_DIV_LEN)) { // LP정보 return 0; } else if (!memcmp(pRcvData, I8, DATA_DIV_LEN)) { // 공매도 return 0; } else if (!memcmp(pRcvData, M9, DATA_DIV_LEN)) { // 회원사정보 return 0; } else if (!memcmp(pRcvData, O4, DATA_DIV_LEN)) { // Buy-in실행수량 return 0; } else if (!memcmp(pRcvData, B2, DATA_DIV_LEN)) { // 시세종가 리커버리 return 0; } else if (!memcmp(pRcvData, B9, DATA_DIV_LEN)) { // 거래원 return 0; } else if (!memcmp(pRcvData, A9, DATA_DIV_LEN)) { // 자사주 배치정보 return 0; } else if (!memcmp(pRcvData, B0, DATA_DIV_LEN)) { // 자사주 매매정보 return 0; } else if (!memcmp(pRcvData, C0, DATA_DIV_LEN)) { // 업종별 투자자별 return 0; } else if (!memcmp(pRcvData, C3, DATA_DIV_LEN)) { // 프로그램매매 호가 return 0; } else if (!memcmp(pRcvData, C4, DATA_DIV_LEN)) { // 대량체결 return 0; } else if (!memcmp(pRcvData, J0, DATA_DIV_LEN)) { // 프로그램매매 전체집계 return 0; } else if (!memcmp(pRcvData, P0, DATA_DIV_LEN)) { // 프로그램매매 투자자별 매매현황 return 0; } else { _gLog->pLog->putLog(_gLog->pLog, (char *)"data_div error...[%2.2s]\n", pRcvData); return -1; } return 0; }

위의 소스는 requestProc()함수에서 일단 배치(A0)처리만 한다. 이후 필요한 시세데이타 처리를 채워나가면 된다.(위의 데이타 형식은 같은 UDP포트로 수신된다.)

위의 구성도에서 데이타가공 부분을 담당하는것이 make_sise_a0()함수이다.

// A0 : 배치종목 가공 int make_sise_a0(LOG_ST *pLog, char *pRcvData, char *utf8KorName, char *utf8EngName, size_t utfLen, PFLIB_ST *pPfLib) { KOSCOM_STOCK_A0 *pA0 = (KOSCOM_STOCK_A0 *)pRcvData; char isin[sizeof(pA0->isin) + 1]; // 종목코드 char korName[sizeof(pA0->kor_name) + 1]; char engName[sizeof(pA0->eng_name) + 1]; // 종목배치처리 memset(isin, 0x00, sizeof(isin)); memcpy(isin, pA0->isin, sizeof(pA0->isin)); // 종목배치를 전부 수신한 경우의 처리 (isin=999999999999) if (!strcmp(isin, "999999999999")) { return 1; } // 데이타를 가공 // 한글/영문종목명 : ' 가 포함되었으면 ''로 치환함 memset(korName, 0x00, sizeof(korName)); memcpy(korName, pA0->kor_name, sizeof(pA0->kor_name)); pPfLib->pString->replaceStr(korName, "'", "''"); // rtrim pPfLib->pString->rtrim(korName, " \t"); // UTF8로 변환 pPfLib->pString->conv_EUCKR_to_UTF8(korName, strlen(korName), utf8KorName, utfLen); memset(engName, 0x00, sizeof(engName)); memcpy(engName, pA0->eng_name, sizeof(pA0->eng_name)); pPfLib->pString->replaceStr(engName, "'", "''"); // rtrim pPfLib->pString->rtrim(engName, " \t"); // UTF8로 변환 pPfLib->pString->conv_EUCKR_to_UTF8(engName, strlen(engName), utf8EngName, utfLen); return 0; }

수신된 배치데이타를 가공하고자 할때 이함수에서 하면 된다. 여기서는 수신된 종목명은 euc-kr형식이므로 이를 utf-8로 변환한다. 수신데이타의 isin값이 999999999999이면 마지막 데이타를 알려주는 패킷으로 처리하지 않는다. 다만 종목배치를 전체 수신후 처리할 작업이 있으면 처리한다.

if (ret == 1) { // 종목배치를 전부 수신한 경우의 처리 (isin=999999999999) // redis 일련번호초기화 _gLog->putLog(_gLog, __FILE__, __FUNCTION__, __LINE__, LOG_TYPE_INFO, 255, "recv_cnt[%d]:[%17.17s]\n", recv_cnt, pRcvData); if (_gRedisMain->command(_gRedisMain, "SET %s %d", A3_SEQ, 0) < 0) { _gLog->pLog->putLog(_gLog->pLog, (char *)"set stock_a3_sequence error...\n"); } // 영업일설정 if (_gRedisMain->command(_gRedisMain, "SET %s %s", BIZ_DAY, _gBizDay) < 0) { _gLog->pLog->putLog(_gLog->pLog, (char *)"set biz_day error...\n"); } return 0; }

체결데이타를 적재할때 필요한 db의 일련번호를 초기화하고, 영업일정보를 설정한다.

make_sise_a0()함수에서 한글종목명, 영문종목명를 변환해 주면 시세데이타와 가공된 데이타를 (받은시세데이타 + utf-8의 한글종목명 + utf-8의 영문종목명) 형식으로 만들어서 db처리, redis처리를 하는 스레드에 채널을 통해 데이타를 전송한다.

// redis처리 _gRedisChannel->send(_gRedisChannel, data, len); // postgres처리 _gPgChannel->send(_gPgChannel, data, len);

이렇게 구성한 이유는 데이타처리를 스레드로 분산해서 데이타처리를 좀 더 빠르게 하기 위함이다.

// PostgreSQL 스레드 void *pg_thread(void *arg) { int ret; char data[BUFFER_SIZE]; CHANNEL_ST *pChannel = (CHANNEL_ST *)arg; int cnt = 0; while (1) { memset(data, 0x00, sizeof(data)); if (pChannel->recv(pChannel, data) < 0) { fprintf(stderr, "pg_thread channel recv error\n"); continue; } if (!memcmp(data, A0, DATA_DIV_LEN)) { char *pUtf8KorName = data + SZ_KOSCOM_STOCK_A0; char *pUtf8EngName = data + SZ_KOSCOM_STOCK_A0 + NAME_UTF8_SIZE; // 종목배치처리 ret = db_worker_a0(_gLog->pLog, _gSql, (KOSCOM_STOCK_A0 *)data, _gPfLib, pUtf8KorName, pUtf8EngName); if (ret == ERR_TERMINATE) { setExitFlag(); break; } } } } // Redis 스레드 void *redis_thread(void *arg) { int ret; char data[BUFFER_SIZE]; CHANNEL_ST *pChannel = (CHANNEL_ST *)arg; int cnt = 0; while (1) { memset(data, 0x00, sizeof(data)); if (pChannel->recv(pChannel, data) < 0) { fprintf(stderr, "redis_thread channel recv error\n"); continue; } if (!memcmp(data, A0, DATA_DIV_LEN)) { char *pUtf8KorName = data + SZ_KOSCOM_STOCK_A0; char *pUtf8EngName = data + SZ_KOSCOM_STOCK_A0 + NAME_UTF8_SIZE; // 종목배치처리 ret = redis_worker_a0(_gLog->pLog, _gRedis, (KOSCOM_STOCK_A0 *)data, _gPfLib, pUtf8KorName, pUtf8EngName); if (ret == ERR_TERMINATE) { setExitFlag(); break; } } } }

메인에서 스레드 생성시 넘겨받은 채널로 채널로 데이타가 들어올때를 기다리다가 데이타가 들어오면 처리한다.

pg_thread()에서 PostgreSQL에 종목배치 시세데이타와 utf-8로 변환한 종목명을 저장한다.

// db저장 int db_worker_a0(LOG_ST *pLog, APSQL_ST *pDbCon, KOSCOM_STOCK_A0 *pA0, PFLIB_ST *pPfLib, char *utf8KorName, char *utf8EngName) { char sqlBuf[1024 * 16]; char typeBuf[128]; memset(typeBuf, 0x00, sizeof(typeBuf)); snprintf(typeBuf, sizeof(typeBuf), pDbCon->db_config.db_type == DB_POSTGRES ? "ON CONFLICT (isin) DO UPDATE SET \n" : "ON DUPLICATE KEY UPDATE \n"); // fprintf(stderr, "db_worker: [%s]\n", typeBuf); memset(sqlBuf, 0x00, sizeof(sqlBuf)); snprintf( sqlBuf, sizeof(sqlBuf), (char *)"INSERT INTO stock_a0 ( \ data_div, \ info_div, \ mkt_div, \ isin, \ serial_num, \ short_code, \ kor_name, \ eng_name, \ biz_day, \ info_group_num, \ stock_grp_id, \ unit_exec, \ drop_code, \ face_chg_code, \ mkt_price_stat, \ reval_reason, \ base_price_chg,\ term_trig_code, \ mkt_warn_stat, \ mkt_warn_code, \ gov_superior, \ mgmt_stat, \ unfaith_stat, \ bypass_stat, \ trd_susp_stat, \ idx_lrg_div, \ idx_mid_div, \ idx_sml_div, \ ind_id, \ kospi200_sec, \ mkt_cap_code, \ mfg_biz, \ krx100_stat, \ div_code, \ inv_eq_div, \ kospi_stat, \ kospi100_stat, \ kospi50_stat, \ base_price, \ prev_close_div, \ prev_close, \ prev_exec_qty, \ prev_trd_amt, \ upper_limit, \ lower_limit, \ sub_price, \ face_price, \ issue_price, \ list_date, \ list_shares, \ settle_trd_stat, \ eps_sign, \ eps, \ per_sign, \ per, \ eps_excl_stat, \ bps_sign, \ bps, \ pbr_sign, \ pbr, \ pbr_excl_stat, \ loss_stat, \ div_share, \ div_excl_stat, \ div_yield, \ start_date, \ end_date, \ event_start, \ event_end, \ exer_price, \ capital, \ credit_stat, \ limit_cond, \ mkt_cond, \ cond_limit_cond, \ fav_limit_cond, \ pri_limit_cond, \ incr_div_code, \ stock_type, \ nat_stock_stat, \ eval_price, \ low_ask, \ high_ask, \ reg_trd_unit, \ ovr_trd_unit, \ reits_type, \ purp_stock_isin, \ curr_code, \ cntry_code, \ mkt_make_stat, \ ovr_trd_stat, \ ovr_close_stat, \ ovr_mass_stat, \ ovr_basket_stat, \ est_exec_stat, \ short_sale_stat, \ etf_mult_sign, \ etf_mult, \ reg_s_stat, \ spac_stat, \ tax_type, \ sub_price_ratio, \ invest_attn_stat, \ delist_date, \ short_overheat, \ etf_copy_code, \ advisor_num, \ exp_date, \ dist_type_code, \ red_start_date, \ red_end_date, \ etf_div_code, \ idx_calc_code, \ idx_mkt_id, \ idx_serial, \ track_idx_code, \ ref_idx_code, \ idx_asset_id1, \ idx_asset_id2, \ lp_order_stat, \ kosdaq150_stat, \ low_liq_stat, \ strange_run_stat, \ krx300_stat, \ upper_qty, \ kospi200_comm_stat, \ invest_warn_stat \ ) \n \ VALUES(\ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ TRIM('%s'), \ TRIM('%s'), \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ %d, \ '%.*s', \ %d, \ %llu, \ %llu, \ %d, \ %d, \ %d, \ %f, \ %d, \ '%.*s', \ %llu, \ '%.*s', \ '%.*s', \ %f, \ '%.*s', \ %f, \ '%.*s', \ '%.*s', \ %f, \ '%.*s', \ %f, \ '%.*s', \ '%.*s', \ %llu, \ '%.*s', \ %f, \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ %f, \ %f, \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ %d, \ %d, \ %d, \ %d, \ %d, \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ %f, \ '%.*s', \ '%.*s', \ '%.*s', \ %f, \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ '%.*s', \ %llu, \ '%.*s', \ '%.*s' \ )\n \ %s \ data_div='%.*s', \ info_div='%.*s', \ mkt_div='%.*s', \ serial_num='%.*s', \ short_code=TRIM('%.*s'), \ kor_name=TRIM('%s'), \ eng_name=TRIM('%s'), \ biz_day='%.*s', \ info_group_num='%.*s', \ stock_grp_id='%.*s', \ unit_exec='%.*s', \ drop_code='%.*s', \ face_chg_code='%.*s', \ mkt_price_stat='%.*s', \ reval_reason='%.*s', \ base_price_chg='%.*s',\ term_trig_code='%.*s', \ mkt_warn_stat='%.*s', \ mkt_warn_code='%.*s', \ gov_superior='%.*s', \ mgmt_stat='%.*s', \ unfaith_stat='%.*s', \ bypass_stat='%.*s', \ trd_susp_stat='%.*s', \ idx_lrg_div='%.*s', \ idx_mid_div='%.*s', \ idx_sml_div='%.*s', \ ind_id='%.*s', \ kospi200_sec='%.*s', \ mkt_cap_code='%.*s', \ mfg_biz='%.*s', \ krx100_stat='%.*s', \ div_code='%.*s', \ inv_eq_div='%.*s', \ kospi_stat='%.*s', \ kospi100_stat='%.*s', \ kospi50_stat='%.*s', \ base_price=%d, \ prev_close_div='%.*s', \ prev_close=%d, \ prev_exec_qty=%llu, \ prev_trd_amt=%llu, \ upper_limit=%d, \ lower_limit=%d, \ sub_price=%d, \ face_price=%f, \ issue_price=%d, \ list_date='%.*s', \ list_shares=%llu, \ settle_trd_stat='%.*s', \ eps_sign='%.*s', \ eps=%f, \ per_sign='%.*s', \ per=%f, \ eps_excl_stat='%.*s', \ bps_sign='%.*s', \ bps=%f, \ pbr_sign='%.*s', \ pbr=%f, \ pbr_excl_stat='%.*s', \ loss_stat='%.*s', \ div_share=%llu, \ div_excl_stat='%.*s', \ div_yield=%f, \ start_date='%.*s', \ end_date='%.*s', \ event_start='%.*s', \ event_end='%.*s', \ exer_price=%f, \ capital=%f, \ credit_stat='%.*s', \ limit_cond='%.*s', \ mkt_cond='%.*s', \ cond_limit_cond='%.*s', \ fav_limit_cond='%.*s', \ pri_limit_cond='%.*s', \ incr_div_code='%.*s', \ stock_type='%.*s', \ nat_stock_stat='%.*s', \ eval_price=%d, \ low_ask=%d, \ high_ask=%d, \ reg_trd_unit=%d, \ ovr_trd_unit=%d, \ reits_type='%.*s', \ purp_stock_isin='%.*s', \ curr_code='%.*s', \ cntry_code='%.*s', \ mkt_make_stat='%.*s', \ ovr_trd_stat='%.*s', \ ovr_close_stat='%.*s', \ ovr_mass_stat='%.*s', \ ovr_basket_stat='%.*s', \ est_exec_stat='%.*s', \ short_sale_stat='%.*s', \ etf_mult_sign='%.*s', \ etf_mult=%f, \ reg_s_stat='%.*s', \ spac_stat='%.*s', \ tax_type='%.*s', \ sub_price_ratio=%f, \ invest_attn_stat='%.*s', \ delist_date='%.*s', \ short_overheat='%.*s', \ etf_copy_code='%.*s', \ advisor_num='%.*s', \ exp_date='%.*s', \ dist_type_code='%.*s', \ red_start_date='%.*s', \ red_end_date='%.*s', \ etf_div_code='%.*s', \ idx_calc_code='%.*s', \ idx_mkt_id='%.*s', \ idx_serial='%.*s', \ track_idx_code='%.*s', \ ref_idx_code='%.*s', \ idx_asset_id1='%.*s', \ idx_asset_id2='%.*s', \ lp_order_stat='%.*s', \ kosdaq150_stat='%.*s', \ low_liq_stat='%.*s', \ strange_run_stat='%.*s', \ krx300_stat='%.*s', \ upper_qty=%llu, \ kospi200_comm_stat='%.*s', \ invest_warn_stat='%.*s' ", (int)sizeof(pA0->data_div), pA0->data_div, (int)sizeof(pA0->info_div), pA0->info_div, (int)sizeof(pA0->mkt_div), pA0->mkt_div, (int)sizeof(pA0->isin), pA0->isin, (int)sizeof(pA0->serial_num), pA0->serial_num, (int)sizeof(pA0->short_code), pA0->short_code, utf8KorName, utf8EngName, (int)sizeof(pA0->biz_day), pA0->biz_day, (int)sizeof(pA0->info_group_num), pA0->info_group_num, (int)sizeof(pA0->stock_grp_id), pA0->stock_grp_id, (int)sizeof(pA0->unit_exec), pA0->unit_exec, (int)sizeof(pA0->drop_code), pA0->drop_code, (int)sizeof(pA0->face_chg_code), pA0->face_chg_code, (int)sizeof(pA0->mkt_price_stat), pA0->mkt_price_stat, (int)sizeof(pA0->reval_reason), pA0->reval_reason, (int)sizeof(pA0->base_price_chg), pA0->base_price_chg, (int)sizeof(pA0->term_trig_code), pA0->term_trig_code, (int)sizeof(pA0->mkt_warn_stat), pA0->mkt_warn_stat, (int)sizeof(pA0->mkt_warn_code), pA0->mkt_warn_code, (int)sizeof(pA0->gov_superior), pA0->gov_superior, (int)sizeof(pA0->mgmt_stat), pA0->mgmt_stat, (int)sizeof(pA0->unfaith_stat), pA0->unfaith_stat, (int)sizeof(pA0->bypass_stat), pA0->bypass_stat, (int)sizeof(pA0->trd_susp_stat), pA0->trd_susp_stat, (int)sizeof(pA0->idx_lrg_div), pA0->idx_lrg_div, (int)sizeof(pA0->idx_mid_div), pA0->idx_mid_div, (int)sizeof(pA0->idx_sml_div), pA0->idx_sml_div, (int)sizeof(pA0->ind_id), pA0->ind_id, (int)sizeof(pA0->kospi200_sec), pA0->kospi200_sec, (int)sizeof(pA0->mkt_cap_code), pA0->mkt_cap_code, (int)sizeof(pA0->mfg_biz), pA0->mfg_biz, (int)sizeof(pA0->krx100_stat), pA0->krx100_stat, (int)sizeof(pA0->div_code), pA0->div_code, (int)sizeof(pA0->inv_eq_div), pA0->inv_eq_div, (int)sizeof(pA0->kospi_stat), pA0->kospi_stat, (int)sizeof(pA0->kospi100_stat), pA0->kospi100_stat, (int)sizeof(pA0->kospi50_stat), pA0->kospi50_stat, pPfLib->pString->strToInt(pA0->base_price, sizeof(pA0->base_price)), (int)sizeof(pA0->prev_close_div), pA0->prev_close_div, pPfLib->pString->strToInt(pA0->prev_close, sizeof(pA0->prev_close)), pPfLib->pString->strToLongLong(pA0->prev_exec_qty, sizeof(pA0->prev_exec_qty)), pPfLib->pString->strToLongLong(pA0->prev_trd_amt, sizeof(pA0->prev_trd_amt)), pPfLib->pString->strToInt(pA0->upper_limit, sizeof(pA0->upper_limit)), pPfLib->pString->strToInt(pA0->lower_limit, sizeof(pA0->lower_limit)), pPfLib->pString->strToInt(pA0->sub_price, sizeof(pA0->sub_price)), pPfLib->pString->strToDouble(pA0->face_price, sizeof(pA0->face_price)), pPfLib->pString->strToInt(pA0->issue_price, sizeof(pA0->issue_price)), (int)sizeof(pA0->list_date), pA0->list_date, pPfLib->pString->strToLongLong(pA0->list_shares, sizeof(pA0->list_shares)), (int)sizeof(pA0->settle_trd_stat), pA0->settle_trd_stat, (int)sizeof(pA0->eps_sign), pA0->eps_sign, pPfLib->pString->strToDouble(pA0->eps, sizeof(pA0->eps)), (int)sizeof(pA0->per_sign), pA0->per_sign, pPfLib->pString->strToDouble(pA0->per, sizeof(pA0->per)), (int)sizeof(pA0->eps_excl_stat), pA0->eps_excl_stat, (int)sizeof(pA0->bps_sign), pA0->bps_sign, pPfLib->pString->strToDouble(pA0->bps, sizeof(pA0->bps)), (int)sizeof(pA0->pbr_sign), pA0->pbr_sign, pPfLib->pString->strToDouble(pA0->pbr, sizeof(pA0->pbr)), (int)sizeof(pA0->pbr_excl_stat), pA0->pbr_excl_stat, (int)sizeof(pA0->loss_stat), pA0->loss_stat, pPfLib->pString->strToLongLong(pA0->div_share, sizeof(pA0->div_share)), (int)sizeof(pA0->div_excl_stat), pA0->div_excl_stat, pPfLib->pString->strToDouble(pA0->div_yield, sizeof(pA0->div_yield)), (int)sizeof(pA0->start_date), pA0->start_date, (int)sizeof(pA0->end_date), pA0->end_date, (int)sizeof(pA0->event_start), pA0->event_start, (int)sizeof(pA0->event_end), pA0->event_end, pPfLib->pString->strToDouble(pA0->exer_price, sizeof(pA0->exer_price)), pPfLib->pString->strToDouble(pA0->capital, sizeof(pA0->capital)), (int)sizeof(pA0->credit_stat), pA0->credit_stat, (int)sizeof(pA0->limit_cond), pA0->limit_cond, (int)sizeof(pA0->mkt_cond), pA0->mkt_cond, (int)sizeof(pA0->cond_limit_cond), pA0->cond_limit_cond, (int)sizeof(pA0->fav_limit_cond), pA0->fav_limit_cond, (int)sizeof(pA0->pri_limit_cond), pA0->pri_limit_cond, (int)sizeof(pA0->incr_div_code), pA0->incr_div_code, (int)sizeof(pA0->stock_type), pA0->stock_type, (int)sizeof(pA0->nat_stock_stat), pA0->nat_stock_stat, pPfLib->pString->strToInt(pA0->eval_price, sizeof(pA0->eval_price)), pPfLib->pString->strToInt(pA0->low_ask, sizeof(pA0->low_ask)), pPfLib->pString->strToInt(pA0->high_ask, sizeof(pA0->high_ask)), pPfLib->pString->strToInt(pA0->reg_trd_unit, sizeof(pA0->reg_trd_unit)), pPfLib->pString->strToInt(pA0->ovr_trd_unit, sizeof(pA0->ovr_trd_unit)), (int)sizeof(pA0->reits_type), pA0->reits_type, (int)sizeof(pA0->purp_stock_isin), pA0->purp_stock_isin, (int)sizeof(pA0->curr_code), pA0->curr_code, (int)sizeof(pA0->cntry_code), pA0->cntry_code, (int)sizeof(pA0->mkt_make_stat), pA0->mkt_make_stat, (int)sizeof(pA0->ovr_trd_stat), pA0->ovr_trd_stat, (int)sizeof(pA0->ovr_close_stat), pA0->ovr_close_stat, (int)sizeof(pA0->ovr_mass_stat), pA0->ovr_mass_stat, (int)sizeof(pA0->ovr_basket_stat), pA0->ovr_basket_stat, (int)sizeof(pA0->est_exec_stat), pA0->est_exec_stat, (int)sizeof(pA0->short_sale_stat), pA0->short_sale_stat, (int)sizeof(pA0->etf_mult_sign), pA0->etf_mult_sign, pPfLib->pString->strToDouble(pA0->etf_mult, sizeof(pA0->etf_mult)), (int)sizeof(pA0->reg_s_stat), pA0->reg_s_stat, (int)sizeof(pA0->spac_stat), pA0->spac_stat, (int)sizeof(pA0->tax_type), pA0->tax_type, pPfLib->pString->strToDouble(pA0->sub_price_ratio, sizeof(pA0->sub_price_ratio)), (int)sizeof(pA0->invest_attn_stat), pA0->invest_attn_stat, (int)sizeof(pA0->delist_date), pA0->delist_date, (int)sizeof(pA0->short_overheat), pA0->short_overheat, (int)sizeof(pA0->etf_copy_code), pA0->etf_copy_code, (int)sizeof(pA0->advisor_num), pA0->advisor_num, (int)sizeof(pA0->exp_date), pA0->exp_date, (int)sizeof(pA0->dist_type_code), pA0->dist_type_code, (int)sizeof(pA0->red_start_date), pA0->red_start_date, (int)sizeof(pA0->red_end_date), pA0->red_end_date, (int)sizeof(pA0->etf_div_code), pA0->etf_div_code, (int)sizeof(pA0->idx_calc_code), pA0->idx_calc_code, (int)sizeof(pA0->idx_mkt_id), pA0->idx_mkt_id, (int)sizeof(pA0->idx_serial), pA0->idx_serial, (int)sizeof(pA0->track_idx_code), pA0->track_idx_code, (int)sizeof(pA0->ref_idx_code), pA0->ref_idx_code, (int)sizeof(pA0->idx_asset_id1), pA0->idx_asset_id1, (int)sizeof(pA0->idx_asset_id2), pA0->idx_asset_id2, (int)sizeof(pA0->lp_order_stat), pA0->lp_order_stat, (int)sizeof(pA0->kosdaq150_stat), pA0->kosdaq150_stat, (int)sizeof(pA0->low_liq_stat), pA0->low_liq_stat, (int)sizeof(pA0->strange_run_stat), pA0->strange_run_stat, (int)sizeof(pA0->krx300_stat), pA0->krx300_stat, pPfLib->pString->strToLongLong(pA0->upper_qty, sizeof(pA0->upper_qty)), (int)sizeof(pA0->kospi200_comm_stat), pA0->kospi200_comm_stat, (int)sizeof(pA0->invest_warn_stat), pA0->invest_warn_stat, typeBuf, (int)sizeof(pA0->data_div), pA0->data_div, (int)sizeof(pA0->info_div), pA0->info_div, (int)sizeof(pA0->mkt_div), pA0->mkt_div, (int)sizeof(pA0->serial_num), pA0->serial_num, (int)sizeof(pA0->short_code), pA0->short_code, utf8KorName, utf8EngName, (int)sizeof(pA0->biz_day), pA0->biz_day, (int)sizeof(pA0->info_group_num), pA0->info_group_num, (int)sizeof(pA0->stock_grp_id), pA0->stock_grp_id, (int)sizeof(pA0->unit_exec), pA0->unit_exec, (int)sizeof(pA0->drop_code), pA0->drop_code, (int)sizeof(pA0->face_chg_code), pA0->face_chg_code, (int)sizeof(pA0->mkt_price_stat), pA0->mkt_price_stat, (int)sizeof(pA0->reval_reason), pA0->reval_reason, (int)sizeof(pA0->base_price_chg), pA0->base_price_chg, (int)sizeof(pA0->term_trig_code), pA0->term_trig_code, (int)sizeof(pA0->mkt_warn_stat), pA0->mkt_warn_stat, (int)sizeof(pA0->mkt_warn_code), pA0->mkt_warn_code, (int)sizeof(pA0->gov_superior), pA0->gov_superior, (int)sizeof(pA0->mgmt_stat), pA0->mgmt_stat, (int)sizeof(pA0->unfaith_stat), pA0->unfaith_stat, (int)sizeof(pA0->bypass_stat), pA0->bypass_stat, (int)sizeof(pA0->trd_susp_stat), pA0->trd_susp_stat, (int)sizeof(pA0->idx_lrg_div), pA0->idx_lrg_div, (int)sizeof(pA0->idx_mid_div), pA0->idx_mid_div, (int)sizeof(pA0->idx_sml_div), pA0->idx_sml_div, (int)sizeof(pA0->ind_id), pA0->ind_id, (int)sizeof(pA0->kospi200_sec), pA0->kospi200_sec, (int)sizeof(pA0->mkt_cap_code), pA0->mkt_cap_code, (int)sizeof(pA0->mfg_biz), pA0->mfg_biz, (int)sizeof(pA0->krx100_stat), pA0->krx100_stat, (int)sizeof(pA0->div_code), pA0->div_code, (int)sizeof(pA0->inv_eq_div), pA0->inv_eq_div, (int)sizeof(pA0->kospi_stat), pA0->kospi_stat, (int)sizeof(pA0->kospi100_stat), pA0->kospi100_stat, (int)sizeof(pA0->kospi50_stat), pA0->kospi50_stat, pPfLib->pString->strToInt(pA0->base_price, sizeof(pA0->base_price)), (int)sizeof(pA0->prev_close_div), pA0->prev_close_div, pPfLib->pString->strToInt(pA0->prev_close, sizeof(pA0->prev_close)), pPfLib->pString->strToLongLong(pA0->prev_exec_qty, sizeof(pA0->prev_exec_qty)), pPfLib->pString->strToLongLong(pA0->prev_trd_amt, sizeof(pA0->prev_trd_amt)), pPfLib->pString->strToInt(pA0->upper_limit, sizeof(pA0->upper_limit)), pPfLib->pString->strToInt(pA0->lower_limit, sizeof(pA0->lower_limit)), pPfLib->pString->strToInt(pA0->sub_price, sizeof(pA0->sub_price)), pPfLib->pString->strToDouble(pA0->face_price, sizeof(pA0->face_price)), pPfLib->pString->strToInt(pA0->issue_price, sizeof(pA0->issue_price)), (int)sizeof(pA0->list_date), pA0->list_date, pPfLib->pString->strToLongLong(pA0->list_shares, sizeof(pA0->list_shares)), (int)sizeof(pA0->settle_trd_stat), pA0->settle_trd_stat, (int)sizeof(pA0->eps_sign), pA0->eps_sign, pPfLib->pString->strToDouble(pA0->eps, sizeof(pA0->eps)), (int)sizeof(pA0->per_sign), pA0->per_sign, pPfLib->pString->strToDouble(pA0->per, sizeof(pA0->per)), (int)sizeof(pA0->eps_excl_stat), pA0->eps_excl_stat, (int)sizeof(pA0->bps_sign), pA0->bps_sign, pPfLib->pString->strToDouble(pA0->bps, sizeof(pA0->bps)), (int)sizeof(pA0->pbr_sign), pA0->pbr_sign, pPfLib->pString->strToDouble(pA0->pbr, sizeof(pA0->pbr)), (int)sizeof(pA0->pbr_excl_stat), pA0->pbr_excl_stat, (int)sizeof(pA0->loss_stat), pA0->loss_stat, pPfLib->pString->strToLongLong(pA0->div_share, sizeof(pA0->div_share)), (int)sizeof(pA0->div_excl_stat), pA0->div_excl_stat, pPfLib->pString->strToDouble(pA0->div_yield, sizeof(pA0->div_yield)), (int)sizeof(pA0->start_date), pA0->start_date, (int)sizeof(pA0->end_date), pA0->end_date, (int)sizeof(pA0->event_start), pA0->event_start, (int)sizeof(pA0->event_end), pA0->event_end, pPfLib->pString->strToDouble(pA0->exer_price, sizeof(pA0->exer_price)), pPfLib->pString->strToDouble(pA0->capital, sizeof(pA0->capital)), (int)sizeof(pA0->credit_stat), pA0->credit_stat, (int)sizeof(pA0->limit_cond), pA0->limit_cond, (int)sizeof(pA0->mkt_cond), pA0->mkt_cond, (int)sizeof(pA0->cond_limit_cond), pA0->cond_limit_cond, (int)sizeof(pA0->fav_limit_cond), pA0->fav_limit_cond, (int)sizeof(pA0->pri_limit_cond), pA0->pri_limit_cond, (int)sizeof(pA0->incr_div_code), pA0->incr_div_code, (int)sizeof(pA0->stock_type), pA0->stock_type, (int)sizeof(pA0->nat_stock_stat), pA0->nat_stock_stat, pPfLib->pString->strToInt(pA0->eval_price, sizeof(pA0->eval_price)), pPfLib->pString->strToInt(pA0->low_ask, sizeof(pA0->low_ask)), pPfLib->pString->strToInt(pA0->high_ask, sizeof(pA0->high_ask)), pPfLib->pString->strToInt(pA0->reg_trd_unit, sizeof(pA0->reg_trd_unit)), pPfLib->pString->strToInt(pA0->ovr_trd_unit, sizeof(pA0->ovr_trd_unit)), (int)sizeof(pA0->reits_type), pA0->reits_type, (int)sizeof(pA0->purp_stock_isin), pA0->purp_stock_isin, (int)sizeof(pA0->curr_code), pA0->curr_code, (int)sizeof(pA0->cntry_code), pA0->cntry_code, (int)sizeof(pA0->mkt_make_stat), pA0->mkt_make_stat, (int)sizeof(pA0->ovr_trd_stat), pA0->ovr_trd_stat, (int)sizeof(pA0->ovr_close_stat), pA0->ovr_close_stat, (int)sizeof(pA0->ovr_mass_stat), pA0->ovr_mass_stat, (int)sizeof(pA0->ovr_basket_stat), pA0->ovr_basket_stat, (int)sizeof(pA0->est_exec_stat), pA0->est_exec_stat, (int)sizeof(pA0->short_sale_stat), pA0->short_sale_stat, (int)sizeof(pA0->etf_mult_sign), pA0->etf_mult_sign, pPfLib->pString->strToDouble(pA0->etf_mult, sizeof(pA0->etf_mult)), (int)sizeof(pA0->reg_s_stat), pA0->reg_s_stat, (int)sizeof(pA0->spac_stat), pA0->spac_stat, (int)sizeof(pA0->tax_type), pA0->tax_type, pPfLib->pString->strToDouble(pA0->sub_price_ratio, sizeof(pA0->sub_price_ratio)), (int)sizeof(pA0->invest_attn_stat), pA0->invest_attn_stat, (int)sizeof(pA0->delist_date), pA0->delist_date, (int)sizeof(pA0->short_overheat), pA0->short_overheat, (int)sizeof(pA0->etf_copy_code), pA0->etf_copy_code, (int)sizeof(pA0->advisor_num), pA0->advisor_num, (int)sizeof(pA0->exp_date), pA0->exp_date, (int)sizeof(pA0->dist_type_code), pA0->dist_type_code, (int)sizeof(pA0->red_start_date), pA0->red_start_date, (int)sizeof(pA0->red_end_date), pA0->red_end_date, (int)sizeof(pA0->etf_div_code), pA0->etf_div_code, (int)sizeof(pA0->idx_calc_code), pA0->idx_calc_code, (int)sizeof(pA0->idx_mkt_id), pA0->idx_mkt_id, (int)sizeof(pA0->idx_serial), pA0->idx_serial, (int)sizeof(pA0->track_idx_code), pA0->track_idx_code, (int)sizeof(pA0->ref_idx_code), pA0->ref_idx_code, (int)sizeof(pA0->idx_asset_id1), pA0->idx_asset_id1, (int)sizeof(pA0->idx_asset_id2), pA0->idx_asset_id2, (int)sizeof(pA0->lp_order_stat), pA0->lp_order_stat, (int)sizeof(pA0->kosdaq150_stat), pA0->kosdaq150_stat, (int)sizeof(pA0->low_liq_stat), pA0->low_liq_stat, (int)sizeof(pA0->strange_run_stat), pA0->strange_run_stat, (int)sizeof(pA0->krx300_stat), pA0->krx300_stat, pPfLib->pString->strToLongLong(pA0->upper_qty, sizeof(pA0->upper_qty)), (int)sizeof(pA0->kospi200_comm_stat), pA0->kospi200_comm_stat, (int)sizeof(pA0->invest_warn_stat), pA0->invest_warn_stat); // fprintf(stderr, "%s\n", sqlBuf); if (pDbCon->db_execute(pDbCon, sqlBuf) < 0) { fprintf(stderr, "[%.*s][%s]:[%s]\n", (int)sizeof(pA0->kor_name), pA0->kor_name, utf8KorName, utf8EngName); return -1; } return 0; }

타파스 시스템에서 데이타 처리는 주로 Redis로 처리하게 설계되었다.

redis_thread()에서는 5가지 형태로 데이타를 생성한다. 추후 또 다른 서비스로 인해 추가가 될 수도 있다.

종목배치(A0)데이타는 가장 먼저 들어오는 종목에 대한 마스타이다.

1) isin코드를 키로 종목마스타를 생성한다. 종목마스타에서 해당종목을 구별하기 위한 코드로 isin코드가 있다.

2) 단축코드(short_code)키로 isin코드를 데이타로 생성한다. 주식정보에서 사용자가 사용하는 코드가 단축코드다. 타파스 시스템에서는 isin코드를 키로 사용하기에 클라이언트에서 단축코드로 요청을 하면 이 단축코드로 isin코드를 구하기 위해서 필요하다. --> 단축코드 --> isin코드 --> 해당 데이타 조회

3) 체결마스타를 생성한다. 종목마스타의 마지막을 알리는 데이타는 isin코드(12자리)에 999999999999가 들어와서 끝을 알려준다. 이때 체결마스타, 호가마스타를 미리 생성한다.

4) 호가마스타를 생성한다. 호가데이타가 들어올때 업데이트를 해서 항상 최신의 호가정보를 가진다.

5) 분봉마스타를 생성한다. 체결데이타가 들어올때 1분봉의 데이타를 업데이트하고 1분이 지날때 마다 1분봉을 적재할때 필요한 데이타이다.

int redis_worker_a0(LOG_ST *pLog, APREDIS_ST *pRedis, KOSCOM_STOCK_A0 *pA0, PFLIB_ST *pPfLib, char *utf8KorName, char *utf8EngName) { // shortcode로 isin를 구하기 if (make_short_code_data(pLog, pRedis, pA0, pPfLib) < 0) { pLog->putLog(pLog, (char *)"make_short_code_data error\n"); return -1; } // master data if (make_master_data(pLog, pRedis, pA0, pPfLib, utf8KorName, utf8EngName) < 0) { pLog->putLog(pLog, (char *)"make_master_data error\n"); return -1; } // 체결데이타 생성 if (make_clear_exec_data(pLog, pRedis, pA0, pPfLib) < 0) { pLog->putLog(pLog, (char *)"make_clear_exec_data error\n"); return -1; } // 호가데이타 생성 if (make_clear_hoga_data(pLog, pRedis, pA0, pPfLib) < 0) { pLog->putLog(pLog, (char *)"make_clear_hoga_data error\n"); return -1; } // 분봉데이타 생성 if (make_clear_min_bong(pLog, pRedis, pA0, pPfLib) < 0) { pLog->putLog(pLog, (char *)"make_clear_min_bong error\n"); return -1; } return 0; }

---