make sina,eastmoney block and block_stock works

requirements.txt

@@ -1,4 +1,4 @@
-zvdata == 1.1.8
+zvdata == 1.1.9
 requests == 2.20.1
 SQLAlchemy == 1.2.14
 pandas == 0.24.2

setup.py

@@ -36,7 +36,7 @@
     # For a discussion on single-sourcing the version across setup.py and the
     # project code, see
     # https://packaging.python.org/en/latest/single_source_version.html
-    version='0.7.1',  # Required
+    version='0.7.2',  # Required
 
     # This is a one-line description or tagline of what your project does. This
     # corresponds to the "Summary" metadata field:
@@ -119,7 +119,7 @@
     # For an analysis of "install_requires" vs pip's requirements files see:
     # https://packaging.python.org/en/latest/requirements.html
 
-    install_requires=['zvdata>=1.1.8', 'requests>=2.20.1', 'SQLAlchemy>=1.2.14', 'pandas>=0.24.2',
+    install_requires=['zvdata>=1.1.9', 'requests>=2.20.1', 'SQLAlchemy>=1.2.14', 'pandas>=0.24.2',
                       'arrow>=0.11.0', 'marshmallow >= 3.2.2', 'tzlocal>=1.5.1', 'xlrd>=1.1.0', 'apscheduler>=3.4.0',
                       'jqdatasdk', 'demjson>=2.2.4', 'marshmallow-sqlalchemy>=0.19.0', 'ccxt>=1.17.191',
                       'plotly>=4.1.0', 'simplejson>=3.16.0',

zvt/recorders/eastmoney/meta/china_stock_category_recorder.py

@@ -3,87 +3,106 @@
 import requests
 
 from zvdata.api import df_to_db
-from zvdata.recorder import Recorder
+from zvdata.recorder import Recorder, TimeSeriesDataRecorder
+from zvdata.utils.time_utils import now_pd_timestamp
 from zvdata.utils.utils import json_callback_param
 from zvt.api.common import china_stock_code_to_id
-from zvt.api.quote import get_entities
-from zvt.domain import BlockStock, BlockCategory
-from zvt.domain.meta.stock_meta import Index
+from zvt.domain import BlockStock, BlockCategory, Block
 
 
-class ChinaStockCategoryRecorder(Recorder):
+class EastmoneyChinaBlockRecorder(Recorder):
     provider = 'eastmoney'
-    data_schema = BlockStock
+    data_schema = Block
 
     # 用于抓取行业/概念/地域列表
     category_map_url = {
         BlockCategory.industry: 'https://nufm.dfcfw.com/EM_Finance2014NumericApplication/JS.aspx?type=CT&cmd=C._BKHY&sty=DCRRBKCPAL&st=(ChangePercent)&sr=-1&p=1&ps=200&lvl=&cb=jsonp_F1A61014DE5E45B7A50068EA290BC918&token=4f1862fc3b5e77c150a2b985b12db0fd&_=08766',
         BlockCategory.concept: 'https://nufm.dfcfw.com/EM_Finance2014NumericApplication/JS.aspx?type=CT&cmd=C._BKGN&sty=DCRRBKCPAL&st=(ChangePercent)&sr=-1&p=1&ps=300&lvl=&cb=jsonp_3071689CC1E6486A80027D69E8B33F26&token=4f1862fc3b5e77c150a2b985b12db0fd&_=08251',
-        BlockCategory.area: 'https://nufm.dfcfw.com/EM_Finance2014NumericApplication/JS.aspx?type=CT&cmd=C._BKDY&sty=DCRRBKCPAL&st=(ChangePercent)&sr=-1&p=1&ps=200&lvl=&cb=jsonp_A597D4867B3D4659A203AADE5B3B3AD5&token=4f1862fc3b5e77c150a2b985b12db0fd&_=02443'
+        # BlockCategory.area: 'https://nufm.dfcfw.com/EM_Finance2014NumericApplication/JS.aspx?type=CT&cmd=C._BKDY&sty=DCRRBKCPAL&st=(ChangePercent)&sr=-1&p=1&ps=200&lvl=&cb=jsonp_A597D4867B3D4659A203AADE5B3B3AD5&token=4f1862fc3b5e77c150a2b985b12db0fd&_=02443'
     }
 
-    # 用于抓取行业包含的股票
-    category_stocks_url = 'https://nufm.dfcfw.com/EM_Finance2014NumericApplication/JS.aspx?type=CT&cmd=C.{}{}&sty=SFCOO&st=(Close)&sr=-1&p=1&ps=300&cb=jsonp_B66B5BAA1C1B47B5BB9778045845B947&token=7bc05d0d4c3c22ef9fca8c2a912d779c'
-
-    def __init__(self, batch_size=10, force_update=False, sleeping_time=10) -> None:
-        super().__init__(batch_size, force_update, sleeping_time)
-
-        self.indices = get_entities(session=self.session, entity_type='index', exchanges=['cn'],
-                                    return_type='domain', provider=self.provider)
-        self.index_ids = [index_item.id for index_item in self.indices]
-
     def run(self):
         for category, url in self.category_map_url.items():
             resp = requests.get(url)
             results = json_callback_param(resp.text)
+            the_list = []
             for result in results:
                 items = result.split(',')
                 code = items[1]
                 name = items[2]
-                id = 'index_cn_{}'.format(code)
-                if id in self.index_ids:
-                    continue
-                self.session.add(Index(id=id, entity_id=id, entity_type='index', exchange='cn', code=code, name=name,
-                                       category=category.value))
-            self.session.commit()
-
-        indices = get_entities(session=self.session, entity_type='index',
-                               return_type='domain', filters=[
-                Index.category.in_(
-                    [BlockCategory.concept.value, BlockCategory.industry.value])],
-                               provider=self.provider)
-
-        for index_item in indices:
-            resp = requests.get(self.category_stocks_url.format(index_item.code, '1'))
-            try:
-                results = json_callback_param(resp.text)
-                the_list = []
-                for result in results:
-                    items = result.split(',')
-                    stock_code = items[1]
-                    stock_id = china_stock_code_to_id(stock_code)
-                    index_id = index_item.id
-                    the_list.append({
-                        'id': '{}_{}'.format(index_id, stock_id),
-                        'entity_id': index_id,
-                        'index_id': index_id,
-                        'stock_id': stock_id
-                    })
-                if the_list:
-                    df = pd.DataFrame.from_records(the_list)
-                    df_to_db(data_schema=self.data_schema, df=df, provider=self.provider)
-
-                self.logger.info('finish recording index:{},{}'.format(index_item.category, index_item.name))
-
-            except Exception as e:
-                self.logger.error("error:,resp.text:", e, resp.text)
-            self.sleep()
-
-
-__all__ = ['ChinaStockCategoryRecorder']
+                entity_id = f'block_cn_{code}'
+                the_list.append({
+                    'id': entity_id,
+                    'entity_id': entity_id,
+                    'entity_type': 'block',
+                    'exchange': 'cn',
+                    'code': code,
+                    'name': name,
+                    'category': category.value
+                })
+            if the_list:
+                df = pd.DataFrame.from_records(the_list)
+                df_to_db(data_schema=self.data_schema, df=df, provider=self.provider,
+                         force_update=True)
+            self.logger.info(f"finish record sina blocks:{category.value}")
+
+
+class EastmoneyChinaBlockStockRecorder(TimeSeriesDataRecorder):
+    entity_provider = 'eastmoney'
+    entity_schema = Block
+
+    provider = 'eastmoney'
+    data_schema = BlockStock
+
+    # 用于抓取行业包含的股票
+    category_stocks_url = 'https://nufm.dfcfw.com/EM_Finance2014NumericApplication/JS.aspx?type=CT&cmd=C.{}{}&sty=SFCOO&st=(Close)&sr=-1&p=1&ps=300&cb=jsonp_B66B5BAA1C1B47B5BB9778045845B947&token=7bc05d0d4c3c22ef9fca8c2a912d779c'
+
+    def __init__(self, exchanges=None, entity_ids=None, codes=None, batch_size=10,
+                 force_update=False, sleeping_time=5, default_size=2000, real_time=False, fix_duplicate_way='add',
+                 start_timestamp=None, end_timestamp=None, close_hour=0, close_minute=0) -> None:
+        super().__init__('block', exchanges, entity_ids, codes, batch_size, force_update, sleeping_time,
+                         default_size, real_time, fix_duplicate_way, start_timestamp, end_timestamp, close_hour,
+                         close_minute)
+
+    def record(self, entity, start, end, size, timestamps):
+        resp = requests.get(self.category_stocks_url.format(entity.code, '1'))
+        try:
+            results = json_callback_param(resp.text)
+            the_list = []
+            for result in results:
+                items = result.split(',')
+                stock_code = items[1]
+                stock_id = china_stock_code_to_id(stock_code)
+                block_id = entity.id
+
+                the_list.append({
+                    'id': '{}_{}'.format(block_id, stock_id),
+                    'entity_id': block_id,
+                    'entity_type': 'block',
+                    'exchange': entity.exchange,
+                    'code': entity.code,
+                    'name': entity.name,
+                    'timestamp': now_pd_timestamp(),
+                    'stock_id': stock_id,
+                    'stock_code': stock_code,
+                    'stock_name': items[2],
+
+                })
+            if the_list:
+                df = pd.DataFrame.from_records(the_list)
+                df_to_db(data_schema=self.data_schema, df=df, provider=self.provider, force_update=True)
+
+            self.logger.info('finish recording block:{},{}'.format(entity.category, entity.name))
+
+        except Exception as e:
+            self.logger.error("error:,resp.text:", e, resp.text)
+        self.sleep()
+
+
+__all__ = ['EastmoneyChinaBlockRecorder', 'EastmoneyChinaBlockStockRecorder']
 
 if __name__ == '__main__':
     # init_log('china_stock_category.log')
 
-    recorder = ChinaStockCategoryRecorder()
+    recorder = EastmoneyChinaBlockStockRecorder(codes=['BK0727'])
     recorder.run()

zvt/recorders/sina/meta/sina_china_stock_category_recorder.py

@@ -37,7 +37,7 @@ def run(self):
 
             for code in tmp_json:
                 name = tmp_json[code].split(',')[1]
-                entity_id = 'index_cn_{}'.format(code)
+                entity_id = f'block_cn_{code}'
                 the_list.append({
                     'id': entity_id,
                     'entity_id': entity_id,
@@ -52,7 +52,7 @@ def run(self):
                 df_to_db(data_schema=self.data_schema, df=df, provider=self.provider,
                          force_update=True)
 
-            self.logger.info(f"finish record sina block:{category.value}")
+            self.logger.info(f"finish record sina blocks:{category.value}")
 
 
 class SinaChinaBlockStockRecorder(TimeSeriesDataRecorder):

zvt/recorders/sina/money_flow/sina_block_money_flow_recorder.py

@@ -91,5 +91,5 @@ def record(self, entity, start, end, size, timestamps):
 __all__ = ['SinaBlockMoneyFlowRecorder']
 
 if __name__ == '__main__':
-    SinaBlockMoneyFlowRecorder(codes=['gn_5Ggn']).run()
+    SinaBlockMoneyFlowRecorder(codes=['new_fjzz']).run()
     # SinaIndexMoneyFlowRecorder().run()

zvt/recorders/sina/money_flow/sina_stock_money_flow_recorder.py

@@ -73,6 +73,7 @@ def record(self, entity, start, end, size, timestamps):
 
             result = {
                 'timestamp': to_pd_timestamp(item['opendate']),
+                'name': entity.name,
                 'close': to_float(item['trade']),
                 'change_pct': to_float(item['changeratio']),
                 'turnover_rate': to_float(item['turnover']) / 10000,