feat: async

pytdx/async/async_base_socket_client.py

@@ -36,10 +36,9 @@ async def connect(self):
     def disconnect(self):
         self.writer.close()
 
-    @asyncio.coroutine
-    def send(self, data, flags=None):
+    async def send(self, data, flags=None):
         if not (self.reader and self.writer):
-            yield from self.connect()
+            await self.connect()
         nsended = len(data)
         self.writer.write(data)
         # yield from self.writer.drain()
@@ -49,11 +48,10 @@ def send(self, data, flags=None):
         self.send_pkg_bytes += nsended
         return nsended
 
-    @asyncio.coroutine
-    def recv(self, buffersize, flags=None):
+    async def recv(self, buffersize, flags=None):
         if not (self.reader and self.writer):
-            yield from self.connect()
-        head_buf = yield from self.reader.read(buffersize)
+            await self.connect()
+        head_buf = await self.reader.read(buffersize)
         self.recv_pkg_num += 1
         self.recv_pkg_bytes += buffersize
         return head_buf

pytdx/async/hq.py

@@ -27,17 +27,17 @@
 
 def exec_command(func):
     @wraps(func)
-    def wrapper(self, *args, **kwargs):
-        connection = self.pool.get_connection()
+    async def wrapper(self, *args, **kwargs):
+        connection = await self.pool.get_connection()
 
         if not connection.connected:
-            yield from make_async_parser(SetupCmd1, connection).call_api()
+            await make_async_parser(SetupCmd1, connection).call_api()
 
-            yield from make_async_parser(SetupCmd2, connection).call_api()
+            await make_async_parser(SetupCmd2, connection).call_api()
 
-            yield from make_async_parser(SetupCmd3, connection).call_api()
+            await make_async_parser(SetupCmd3, connection).call_api()
 
-        data = yield from func(self, *args, **kwargs, connection=connection)
+        data = await func(self, *args, **kwargs, connection=connection)
         self.pool.release(connection)
         return data
 

pytdx/async/pool.py

@@ -2,16 +2,16 @@
 import asyncio
 from itertools import chain
 from .async_base_socket_client import AsyncTrafficStatSocket
-
+import pandas as pd
 import os
 
 
 class ConnectionPool(object):
 
-    def __init__(self, ip, port, max_connections=None, loop=None):
+    def __init__(self, ip, port, max_connections=300, loop=None):
 
         self.pid = os.getpid()
-        self.max_connections = max_connections or 2 ** 31
+        self.max_connections = max_connections
 
         self.loop = loop or asyncio.get_event_loop()
         self.ip = ip
@@ -20,8 +20,12 @@ def __init__(self, ip, port, max_connections=None, loop=None):
         self.created_connect = 0
         self._in_use_connections = set()
 
-    def get_connection(self):
+    async def get_connection(self):
         try:
+            if self.created_connect >= self.max_connections:
+                # if self.created_connect > self.max_connections:
+                while len(self._available_connections) == 0:
+                    await asyncio.sleep(0.2)
             connection = self._available_connections.pop()
         except IndexError:
             connection = self.make_connection()
@@ -30,9 +34,6 @@ def get_connection(self):
         return connection
 
     def make_connection(self):
-        if self.created_connect >= self.max_connections:
-            raise ConnectionError("Too many connections")
-
         self.created_connect += 1
         return AsyncTrafficStatSocket(self.ip, self.port, self.loop)
 
@@ -47,5 +48,5 @@ def disconnect(self):
         for connection in all_conns:
             connection.disconnect()
 
-    def run_until_complete(self,*args,**kwargs):
-        return self.loop.run_until_complete(*args,**kwargs)
+    def run_until_complete(self, *args, **kwargs):
+        return self.loop.run_until_complete(*args, **kwargs)

pytdx/async/reflection.py

@@ -27,29 +27,27 @@ def make_async_parser(parser, connection):
     :return:
     """
 
-    @asyncio.coroutine
-    def call_api(self):
+    async def call_api(self):
         if self.lock:
             with self.lock:
                 log.debug("sending thread lock api call")
-                result = yield from self._call_api()
+                result = await self._call_api()
         else:
-            result = yield from self._call_api()
+            result = await self._call_api()
         return result
 
-    @asyncio.coroutine
-    def _call_api(self):
+    async def _call_api(self):
         if not self.send_pkg:
             SendPkgNotReady("send pkg not ready")
 
-        yield from connection.send(self.send_pkg)
-        head_buf = yield from connection.recv(self.rsp_header_len)
+        await connection.send(self.send_pkg)
+        head_buf = await connection.recv(self.rsp_header_len)
         if len(head_buf) == self.rsp_header_len:
             _, _, _, zipsize, unzipsize = struct.unpack("<IIIHH", head_buf)
             body_buf = bytearray()
 
             while True:
-                buf = yield from connection.recv(zipsize)
+                buf = await connection.recv(zipsize)
                 len_buf = len(buf)
                 body_buf.extend(buf)
                 if not (buf) or len_buf == 0 or len(body_buf) == zipsize:
@@ -82,40 +80,41 @@ def _call_api(self):
     from pytdx.hq import TdxHq_API
     import pprint
 
+
     def time_async():
 
         pool = ConnectionPool(ip='101.227.73.20', port=7709)
 
-        def exec_command(pool,cmd):
-
+        async def exec_command(pool, cmd):
             connection = pool.get_connection()
 
             if not connection.connected:
-                yield from make_async_parser(SetupCmd1, connection).call_api()
-
-                yield from  make_async_parser(SetupCmd2, connection).call_api()
+                await make_async_parser(SetupCmd1, connection).call_api()
 
-                yield from make_async_parser(SetupCmd3, connection).call_api()
+                await  make_async_parser(SetupCmd2, connection).call_api()
 
+                await make_async_parser(SetupCmd3, connection).call_api()
 
             async_cmd = make_async_parser(cmd, connection)
 
             async_cmd.setParams(8, 0, '000001', 0, 80)
 
-            data = yield from async_cmd.call_api()
+            data = await async_cmd.call_api()
             pool.release(connection)
 
             return data
 
-        res = [exec_command(pool,GetSecurityBarsCmd) for i in range(100)]
+        res = [exec_command(pool, GetSecurityBarsCmd) for i in range(100)]
         pool.run_until_complete(asyncio.wait(res))
 
+
     def time_orig():
         api = TdxHq_API()
         api.connect(ip='218.108.98.244', port=7709)
 
         for i in range(100):
             api.get_security_bars(8, 0, '000001', 0, 80)
 
+
     # print(timeit.timeit(time_async, number=1))
-    print(timeit.timeit(time_orig, number=1))
+    print(timeit.timeit(time_orig, number=1))