Examples

[rocky-d]

Examples can be shared here.

[rocky-d]

"""
Alpha Template:

group_rank(<fundamental6>/cap, subindustry)
"""

import asyncio
import wqb


async def main() -> None:
    wqb_auth = '<email>', '<password>'
    logger = wqb.wqb_logger()
    wqbs = wqb.WQBSession(wqb_auth, logger=logger)

    resp = wqbs.locate_field('open')
    print(f"{resp.ok = }")

    fields = []
    for resp in wqbs.search_fields('USA', 1, 'TOP3000', dataset_id='fundamental6'):
        fields.extend(item['id'] for item in resp.json()['results'])
    print(f"{len(fields) = }")

    alphas = []
    for field in fields:
        alpha = {
            'type': 'REGULAR',
            'settings': {
                'instrumentType': 'EQUITY',
                'region': 'USA',
                'universe': 'TOP3000',
                'delay': 1,
                'decay': 0,
                'neutralization': 'SUBINDUSTRY',
                'truncation': 0.08,
                'pasteurization': 'ON',
                'unitHandling': 'VERIFY',
                'nanHandling': 'ON',
                'language': 'FASTEXPR',
                'visualization': False,
            },
            'regular': f"group_rank({field}/cap, subindustry)",
        }
        alphas.append(alpha)
    print(f"{len(alphas) = }")

    resps = await wqbs.concurrent_simulate(alphas, 3, log_gap=10)
    print('alpha_ids:', *(resp.json()['alpha'] for resp in resps), sep='\n')


if __name__ == '__main__':
    asyncio.run(main())

[rocky-d]

"""
An example of calling `patch_properties(...)` immediately once a `simulate(...)` coroutine is done.
"""

import asyncio
import wqb


async def main() -> None:
    wqbs = wqb.WQBSession(('<email>', '<password>'), logger=wqb.wqb_logger())

    fields = []
    for resp in wqbs.search_fields('USA', 1, 'TOP3000', dataset_id='fundamental6'):
        fields.extend(item['id'] for item in resp.json()['results'])
    alphas = []
    for field in fields:
        alpha = {
            'type': 'REGULAR',
            'settings': {
                'instrumentType': 'EQUITY',
                'region': 'USA',
                'universe': 'TOP3000',
                'delay': 1,
                'decay': 0,
                'neutralization': 'SUBINDUSTRY',
                'truncation': 0.08,
                'pasteurization': 'ON',
                'unitHandling': 'VERIFY',
                'nanHandling': 'ON',
                'language': 'FASTEXPR',
                'visualization': False,
            },
            'regular': f"group_rank({field}/cap, subindustry)",
        }
        alphas.append(alpha)
    print(f"{len(alphas) = }")

    def on_success(vars: dict[str, object]) -> None:
        wqbs: wqb.WQBSession = vars['self']
        for child in vars['resp'].json()['children']:
            resp = wqbs.get('https://api.worldquantbrain.com/simulations/' + child)
            alpha_id = resp.json()['alpha']
            resp = wqbs.patch_properties(
                alpha_id,
                # favorite=False,  # False, True
                # hidden=False,  # False, True
                # name=NULL,  # '<name>'
                # category=NULL,  # 'ANALYST', 'FUNDAMENTAL'
                # tags=NULL,  # '<tag>', ['tag_0', 'tag_1', 'tag_2']
                # color=NULL,  # 'RED', 'YELLOW', 'GREEN', 'BLUE', 'PURPLE'
                # regular_description=NULL,  # '<regular_description>'
                log=None,
            )

    multi_alphas = wqb.to_multi_alphas(alphas, 10)
    concurrency = 10
    resps = await wqbs.concurrent_simulate(
        multi_alphas,
        concurrency,
        on_success=on_success,
    )


if __name__ == '__main__':
    asyncio.run(main())

[rocky-d]

首先on_success参数不是必须的，如果你需要用到on_success实现一些回调操作，比如设置color属性，并且是用户权限，可以参考下面的on_success函数写法：

    def on_success(vars: dict[str, object]) -> None:
        resp = vars['resp']
        alpha_id = resp.json()['alpha']
        resp = wqbs.patch_properties(
            alpha_id,
            # favorite=False,  # False, True
            # hidden=False,  # False, True
            # name=NULL,  # '<name>'
            # category=NULL,  # 'ANALYST', 'FUNDAMENTAL'
            # tags=NULL,  # '<tag>', ['tag_0', 'tag_1', 'tag_2']
            # color=NULL,  # 'RED', 'YELLOW', 'GREEN', 'BLUE', 'PURPLE'
            # regular_description=NULL,  # '<regular_description>'
            log=None,
        )

[ElysianDreamer]

你好！运行的时候会报错如下，请问这是为什么

KeyError Traceback (most recent call last)
File d:\anaconda\envs\WorldQuant\Lib\site-packages\wqb\wqb_session.py:1118, in WQBSession.retry(self, method, url, max_tries, max_key_errors, max_value_errors, delay_key_error, delay_value_error, on_start, on_finish, on_success, on_failure, log, *args, **kwargs)
1117 try:
-> 1118 await asyncio.sleep(float(resp.headers[RETRY_AFTER]))
1119 except KeyError as e:

File d:\anaconda\envs\WorldQuant\Lib\site-packages\requests\structures.py:52, in CaseInsensitiveDict.getitem(self, key)
51 def getitem(self, key):
---> 52 return self._store[key.lower()][1]

KeyError: 'retry-after'

During handling of the above exception, another exception occurred:

KeyError Traceback (most recent call last)
Cell In[11], line 62
54 resps = await session.concurrent_simulate(
55 multi_alphas,
56 concurrency,
57 on_success=on_success,
58 )
61 if name == 'main':
---> 62 asyncio.run(main())
...
42 # favorite=False, # False, True
(...) 49 log=None,
50 )

KeyError: 'alpha'用的时候只改了一个地方
for experess in first_order:
alpha = {
'type': 'REGULAR',
'settings': {
'instrumentType': 'EQUITY',
'region': 'ASI',
'universe': 'MINVOL1M',
'delay': 1,
'decay': 0,
'neutralization': 'SUBINDUSTRY',
'truncation': 0.08,
'pasteurization': 'ON',
'unitHandling': 'VERIFY',
'nanHandling': 'ON',
'language': 'FASTEXPR',
'visualization': False,
},
'regular': experess,
}
alphas.append(alpha)
print(f"{len(alphas) = }")

[compegant]

Sorry to interrupt, but does setting alphaname using the on_sucess method affect the speed of concurrent simulation?

[rocky-d]

Sorry to interrupt, but does setting alphaname using the on_sucess method affect the speed of concurrent simulation?

It depends on what your on_success function does.

In this example, it will wait until on_success finished and then continue simulating next Alpha or MultiAlpha. On the whole, the time cost of these single requests is not too much.

If you really care about the efficiency, another implementation shown below can be considered.

    def on_success(vars: dict[str, object]) -> None:
        async def on_success_coro():
            # Use asyncio-based libraries (e.g., aiohttp) to implement the logic.
            pass

        asyncio.create_task(on_success_coro())

[rocky-d]

@ElysianDreamer

可能是没有分清Alpha和MultiAlpha导致的的问题（个人见解仅供参考）。