ES+Kibana实时展示股票收益

「爱情、让人受尽委屈。」 2022-10-09 12:16 227阅读 0赞

### ES+Kibana实时展示股票收益 ###

*  前言
 *  一、初期调研
 *  二、ES基本操作
 *   *  1.创建连接
     *  2.创建Index
     *  3.插入数据
     *   *  3.1普通插入
         *  3.2批量插入
     *  4.查找
     *  5. 删除
 *  三、Kibana基本操作
 *   *  使用SQL查询
     *  各种控件
 *  总结

--------------------

# 前言 #

应用场景：

在项目中需要监控股票的实时收益。本人不懂前端开发，于是想用日志监控套件ELK进行数据录入和展示。因为股票池最多也就全A股，虽然有好几天的持仓，总体数据量不大。

--------------------

# 一、初期调研 #

1.  首先ELK的核心是Elasticsearch，这个东西超级强大，用来做实时站内搜索引擎的。ES用了分片和复制的技术，可以在保证数据高可用的情况下，提供急速并发查询的能力。ES的强项在于读（查询），而非写。
2.  ELK最经典的应用场景是日志搜集与分析预警。跟我个人的使用场景不同。Logstash比较重，消耗资源，暂时放弃这个。
3.  考虑到对数据持久化的要求不高，准备直接调用**python**接口写入ES

# 二、ES基本操作 #

## 1.创建连接 ##

1.  因为ES用的是HTTPS的链接，我们这里需要把证书加上。
2.  因为内网没有用域名，直接用的IP，所以需要加上加上context.check\_hostname = False  
    详细内容参考[官方文档][Link 1]

import datetime
    from elasticsearch import helpers
    from elasticsearch import Elasticsearch
    from elasticsearch_dsl import Search
    from ssl import create_default_context
    
    context = create_default_context(cafile="elastic-2.ca.crt")
    context.check_hostname = False
    es = Elasticsearch("https://127.0.0.1:31920", ssl_context=context, http_auth=('admin', 'admin'))

## 2.创建Index ##

![ES数据结构和mysql的对应关系][ES_mysql]  
ES的index对应的就是关系型数据库，type即将被抛弃，在不使用type的情况下，我们直接可以把它当个表也行。

这里面我们需要搞清楚每个字段的类型。字段的设置会影响查找效率，如果不需要分词什么的，字符串尽量用keyword就可以了。

def create_index(index_name, es):
        if es.indices.exists(index_name):  # 判断是否存在
            return
    
        request_body = {
        
            "mappings": {
        
                "properties": {
        
                    "buy_date": {
        
                        "type": "keyword"
                    },
                    "ticker": {
        
                        "type": "keyword"
                    },
                    "buy_price": {
        
                        "type": "float"
                    },
                    "current_price": {
        
                        "type": "float"
                    },
                    "quantity": {
        
                        "type": "integer"
                    },
                    "Return": {
        
                        "type": "float"
                    },
                    "market_val": {
        
                        "type": "float"
                    },
                    "updateTime": {
        
                        "type": "date",
                        "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
                    }
                }
            }
        }
    
        es.indices.create(index=index_name, body=request_body)

## 3.插入数据 ##

### 3.1普通插入 ###

注意这里面的时间，我是把当前时间转成了UTC0时间，因为**后面Kibana在使用数据的时候，默认这里的时间就是UTC-0的。**

def save_data_to_es(es, index_name='s2t0_test'):
        body_dict = {
        
            "buy_date": "20210620",
            "ticker": '000010.SZ',
            "buy_price": 2.65,
            "current_price": 3.1,
            "quantity": 200,
            "Return": 0.1,
            "market_val": 140,
            "updateTime": datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
            }
        es.index(index=index_name, body=body_dict)

### 3.2批量插入 ###

def bulk_save_to_es(self):
            if len(self.data) == 0:
                return
                
            datas = []
            for item in self.data.values():  # item是需要保持的数据，dict类型
                tmp_d = dict()
                item['updateTime'] = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
                tmp_d['_index'] = self.index_name
                tmp_d["_source"] = item
                datas.append(tmp_d)
    
            helpers.bulk(self.es_client, datas)

## 4.查找 ##

def search_es(es):
        day = "20210622"
        query_d = {
        
            "bool": {
        
                "must": [
                    {
        
                        "match": {
        
                            "ticker": "000009.SZ"
                        }
                    },
                    {
        
                        "match": {
        
                            "buy_date": day
                        }
                    }
                ]
            }
        }
        s = Search(using=es, index='s2t0_test').query(query_d)
        for hit in s:
            print(hit)

## 5. 删除 ##

def delete_by_date(index, es, date):
        query = {
        "query": {
        "match": {
        "buy_date": date}}}
        res = es.delete_by_query(index=index, body=query)
    
    # delete all
    delete_by_query(index=self.index_name, body={
        "query": {
        "match_all": {
        }}})

# 三、Kibana基本操作 #

## 使用SQL查询 ##

在一般的应用场景下，使用sql查询会让学习曲线陡降，特别是结合dev tools，简单快捷优美。

POST /_sql?format=txt
    {
        
      "query": "SELECT * FROM real_time_alpha where date='20210629' order by updateTime desc limit 1 "
    }

## 各种控件 ##

Kibana主要需要学会使用各种控件，当前主要是表格。下面是一些注意事项：

1.  首先需要针对你的index，建立index pattern，要不然后面的dashboard无法搜索到。
2.  为了时间正常显示，在高级设置里，可以选择当前的时区。
3.  可以设置显示的刷新时间，最好符合自己的数据刷新频率。

下面展示的效果:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2duaW5naA_size_16_color_FFFFFF_t_70_pic_center]

# 总结 #

1.  开发过程中可以借用Kibana的Dev tools进行数据的查看和命令的验证，不要直接上代码
2.  ES提供的python接口还是比较简单易用的，但是还是喜欢传json类型的命令或者数据，这样针对简单业务场景下，学习成本更低。
3.  至于Elasticsearch-dsl，个人觉得学习曲线略微陡峭。如果不需要重度依赖，完全没必要去学。
4.  基本上两三天可以从零搭建一个实时显示控件的dashboard，ES+Kibana非常给力！
5.  Elasticsearch**支持SQL查询**，虽然会有点局限，这个对于简单的业务来说简直大杀器

下面是一些参考的资料：

*  [ES基础查询][ES]
 *  [Elasticsearch 7 : Python 客户端操作][Elasticsearch 7 _ Python]
 *  [kibana官方文档][kibana]

[Link 1]: https://elasticsearch-py.readthedocs.io/en/v7.13.2/#tls-ssl-and-authentication
[ES_mysql]: /images/20221005/4c70a1b1b25448e8a7317b0fad9f18cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2duaW5naA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/e95d5d0a61414932bb62727fb37a16b7.png
[ES]: https://www.elastic.co/guide/cn/elasticsearch/guide/current/_most_important_queries.html
[Elasticsearch 7 _ Python]: https://www.letianbiji.com/elasticsearch/es7-python-client.html
[kibana]: https://www.elastic.co/guide/cn/kibana/current/index.html