py, abquant etf list

.gitignore

@@ -26,6 +26,8 @@ wheels/
 .installed.cfg
 *.egg
 
+poetry.lock
+
 # PyInstaller
 #  Usually these files are written by a python script from a template
 #  before PyInstaller builds the exe, so as to inject date/other infos into it.

README.org

@@ -129,6 +129,9 @@ abquant 分为三大模块 a) 数据获取 b) 策略回测 c) 模拟/实盘交
 
        # 股票财务数据
        abquant stock financial
+
+       # ETF 列表获取
+       abquant etf list
        # ...
      #+end_src
 

abquant/cli.py

@@ -13,6 +13,7 @@
     create_stock_info,
     create_stock_financial,
     create_base,
+    create_etf_list,
 )
 
 set_loggers()
@@ -45,6 +46,10 @@ def save_base():
 def stock():
     """Manages stocks."""
 
+@cli.group()
+def etf():
+    """Manages etf."""
+
 
 @stock.command("day")
 @click.option(
@@ -150,8 +155,12 @@ def stock_block():
     """stock block."""
     create_stock_block()
 
-
 @stock.command("financial")
 def stock_financial():
     """stock financial."""
     create_stock_financial()
+
+@etf.command("list")
+def etf_list():
+    """etf list."""
+    create_etf_list()

abquant/data/base.py

@@ -140,3 +140,11 @@ def create_stock_financial():
     broker = get_broker()
     s = broker.Stock()
     s.create_financial()
+
+
+@time_counter
+def create_etf_list():
+    broker = get_broker("tdx")
+    brokers_api = [get_broker_api(b) for b in ["tdx", "ths", "qa"]]
+    e = broker.Etf()
+    e.create_list(brokers_api=brokers_api)

abquant/data/qa_api.py

@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-
 from abquant.utils.code import code_tostr
+from typing import Optional
 import pandas as pd
 
 
-def get_stock_block():
+def get_stock_block() -> Optional[pd.DataFrame]:
     """ths的版块数据
 
     Returns:
@@ -25,5 +26,9 @@ def get_stock_block():
         return None
 
 
+def get_stock_list(type_=None) -> Optional[pd.DataFrame]:
+    return
+
+
 def my_name():
     return "QA"

abquant/data/tdx.py

@@ -23,10 +23,10 @@
 from threading import RLock
 from functools import partial
 import datetime
+import pandas as pd
 import os
 
 
-# class Stock(object):
 class Stock(ISecurity):
     def __init__(self, *args, **kwargs):
         "docstring"
@@ -94,7 +94,10 @@ def saving_work(idx):
                 finish = datetime.datetime.now()
                 interval = (finish - begin).total_seconds()
                 text = "{}, {} -> {}, {:<04.2}s".format(
-                    "{} {}".format(stockOrIndex, code), start_time_plus_one, end_time, interval,
+                    "{} {}".format(stockOrIndex, code),
+                    start_time_plus_one,
+                    end_time,
+                    interval,
                 )
                 for _ in trange(df.size, desc=text):
                     pass
@@ -292,8 +295,10 @@ def saving_work(idx):
             try:
                 begin = datetime.datetime.now()
                 df = get_stock_info(str(code))
-                if df is None or df.empty:
-                    slog.warn("df is empty. {}".format(code))
+                if isinstance(df, pd.DataFrame) and df.empty:
+                    slog.warn("df is empty.")
+                    return
+                if not isinstance(df, pd.DataFrame) and not df:
                     return
                 coll.insert_many(to_json_from_pandas(df))
                 finish = datetime.datetime.now()
@@ -327,12 +332,17 @@ def create_block(self, *args, **kwargs):
         def saving_work(b_api):
             try:
                 begin = datetime.datetime.now()
-                data = b_api.get_stock_block()
-                coll.insert_many(to_json_from_pandas(data))
+                df = b_api.get_stock_block()
+                if isinstance(df, pd.DataFrame) and df.empty:
+                    slog.warn("df is empty.")
+                    return
+                if not isinstance(df, pd.DataFrame) and not df:
+                    return
+                coll.insert_many(to_json_from_pandas(df))
                 finish = datetime.datetime.now()
                 interval = (finish - begin).total_seconds()
                 text = "stock block {}, {:<04.2}s".format(b_api.my_name(), interval)
-                for _ in trange(data.size, desc=text):
+                for _ in trange(df.size, desc=text):
                     pass
             except Exception as e:
                 slog.error("{}".format(e))
@@ -413,3 +423,59 @@ def create_min(self, *args, **kwargs):
         stockOrIndex = kwargs.pop("stockOrIndex", "future")
         slog.debug("{} {} min {}".format(self.getClassName(), stockOrIndex, self.freqs))
         return
+
+
+class Etf(ISecurity):
+    def __init__(self, *args, **kwargs):
+        "docstring"
+        self._client = pymongo.MongoClient(Setting.get_mongo())
+        self._db = self._client[Setting.DBNAME]
+        self.codes = kwargs.get("codes", [])
+        self.freqs = []
+
+    def accept(self, visitor: ISecurityVisitor) -> None:
+        visitor.create_day(self)
+        visitor.create_min(self)
+
+    def create_day(self, *args, **kwargs):
+        stockOrIndex = kwargs.pop("etf", "")
+        slog.debug("{} {} day".format(self.getClassName(), stockOrIndex))
+        return
+
+    def create_min(self, *args, **kwargs):
+        stockOrIndex = kwargs.pop("etf", "")
+        slog.debug("{} {} min {}".format(self.getClassName(), stockOrIndex, self.freqs))
+        return
+
+    def create_list(self, *args, **kwargs):
+        """save etf list
+
+        Keyword Arguments:
+            client {[type]} -- [description] (default: {DATABASE})
+        """
+        brokers_api = kwargs.get("brokers_api", [])
+        self._db.drop_collection("etf_list")
+        coll = self._db.etf_list
+        coll.create_index([("code", pymongo.ASCENDING)])
+
+        def saving_work(b_api):
+            try:
+                begin = datetime.datetime.now()
+                df = b_api.get_stock_list(type_="etf")
+                if isinstance(df, pd.DataFrame) and df.empty:
+                    slog.warn("df is empty.")
+                    return
+                if not isinstance(df, pd.DataFrame) and not df:
+                    return
+                coll.insert_many(to_json_from_pandas(df))
+                finish = datetime.datetime.now()
+                interval = (finish - begin).total_seconds()
+                text = "etf list {}, {:<04.2}s".format(b_api.my_name(), interval)
+                for _ in trange(df.size, desc=text):
+                    pass
+            except Exception as e:
+                slog.error("{}".format(e))
+
+        tqdm.set_lock(RLock())
+        with ThreadPoolExecutor(max_workers=1) as p:
+            p.map(partial(saving_work), brokers_api)

abquant/data/tdx_api.py

@@ -552,7 +552,7 @@ def get_stock_min(code, start, end, frequence="1min", ip=None, port=None):
 
 
 @retry(stop_max_attempt_number=3, wait_random_min=50, wait_random_max=100)
-def QA_fetch_get_stock_latest(code, frequence="day", ip=None, port=None):
+def get_stock_latest(code, frequence="day", ip=None, port=None):
     ip, port = get_mainmarket_ip(ip, port)
     code = [code] if isinstance(code, str) else code
     api = TdxHq_API(multithread=True)
@@ -604,7 +604,7 @@ def QA_fetch_get_stock_latest(code, frequence="day", ip=None, port=None):
 
 
 @retry(stop_max_attempt_number=3, wait_random_min=50, wait_random_max=100)
-def QA_fetch_get_stock_realtime(code=["000001", "000002"], ip=None, port=None):
+def get_stock_realtime(code=["000001", "000002"], ip=None, port=None):
     ip, port = get_mainmarket_ip(ip, port)
     # reversed_bytes9 --> 涨速
     # active1,active2 --> 活跃度
@@ -677,7 +677,7 @@ def QA_fetch_get_stock_realtime(code=["000001", "000002"], ip=None, port=None):
 
 
 @retry(stop_max_attempt_number=3, wait_random_min=50, wait_random_max=100)
-def QA_fetch_get_index_realtime(code=["000001"], ip=None, port=None):
+def get_index_realtime(code=["000001"], ip=None, port=None):
     ip, port = get_mainmarket_ip(ip, port)
     # reversed_bytes9 --> 涨速
     # active1,active2 --> 活跃度
@@ -830,7 +830,7 @@ def get_index_min(code, start, end, frequence="1min", ip=None, port=None):
 
 
 @retry(stop_max_attempt_number=3, wait_random_min=50, wait_random_max=100)
-def QA_fetch_get_index_list(ip=None, port=None):
+def get_index_list(ip=None, port=None):
     """获取指数列表
     Keyword Arguments:
         ip {[type_]} -- [description] (default: {None})

abquant/data/ths_api.py

@@ -1,7 +1,8 @@
+from typing import Optional
 import pandas as pd
 
 
-def get_stock_block():
+def get_stock_block() -> Optional[pd.DataFrame]:
     """ths的版块数据
 
     Returns:
@@ -15,5 +16,9 @@ def get_stock_block():
         return None
 
 
+def get_stock_list(type_=None) -> Optional[pd.DataFrame]:
+    return
+
+
 def my_name():
     return "THS"