API文档

def merge(left, right, by):
    """merge two numpy structured arrays by `by` key

    njit fail if one of left, right contains object, not plain type, but the loop is
    very fast, cost 0.0001 seconds

    Args:
        left ([type]): [description]
        right ([type]): [description]
        by ([type]): [description]

    Returns:
        [type]: [description]
    """
    i, j = 0, 0

    while j < len(right) and i < len(left):
        if right[j][by] < left[by][i]:
            j += 1
        elif right[j][by] == left[by][i]:
            left[i] = right[j]
            i += 1
            j += 1
        else:
            i += 1

    return left

async def calc_checksums(day: datetime.date, codes: List) -> dict:
    """
    Args:
        day:
        codes:

    Returns:
        返回值为以code为键，该证券对应的{周期：checksum}的集合为值的集合
    """
    end_time = arrow.get(day, tzinfo=cfg.tz).replace(hour=15)

    checksums = {}
    for i, code in enumerate(codes):
        try:
            checksum = {}
            d = await cache.get_bars_raw_data(code, day, 1, FrameType.DAY)
            if d:
                checksum[f"{FrameType.DAY.value}"] = xxhash.xxh32_hexdigest(d)
            d = await cache.get_bars_raw_data(code, end_time, 240, FrameType.MIN1)
            if d:
                checksum[f"{FrameType.MIN1.value}"] = xxhash.xxh32_hexdigest(d)

            d = await cache.get_bars_raw_data(code, end_time, 48, FrameType.MIN5)
            if d:
                checksum[f"{FrameType.MIN5.value}"] = xxhash.xxh32_hexdigest(d)

            d = await cache.get_bars_raw_data(code, end_time, 16, FrameType.MIN15)
            if d:
                checksum[f"{FrameType.MIN15.value}"] = xxhash.xxh32_hexdigest(d)

            d = await cache.get_bars_raw_data(code, end_time, 8, FrameType.MIN30)
            if d:
                checksum[f"{FrameType.MIN30.value}"] = xxhash.xxh32_hexdigest(d)

            d = await cache.get_bars_raw_data(code, end_time, 4, FrameType.MIN60)
            if d:
                checksum[f"{FrameType.MIN60.value}"] = xxhash.xxh32_hexdigest(d)

            checksums[code] = checksum
        except Exception as e:
            logger.exception(e)

        if (i + 1) % 500 == 0:
            logger.info("calc checksum progress: %s/%s", i + 1, len(codes))

    return checksums

async def do_validation(secs: List[str] = None, start: str = None, end: str = None):
    """对列表secs中指定的证券行情数据按start到end指定的时间范围进行校验

    Args:
        secs (List[str], optional): [description]. Defaults to None.
        start (str, optional): [description]. Defaults to None.
        end (str, optional): [description]. Defaults to None.

    Returns:
        [type]: [description]
    """
    logger.info("start validation...")
    report = logging.getLogger("validation_report")

    cfg = cfg4py.init(get_config_dir(), False)

    await emit.start(engine=emit.Engine.REDIS, dsn=cfg.redis.dsn, start_server=True)
    await omicron.init()
    start = int(start or await cache.sys.get("jobs.bars_validation.range.start"))
    if end is None:
        end = tf.date2int(arrow.now().date())
    else:
        end = int(end or await cache.sys.get("jobs.bars_validation.range.stop"))

    if secs is None:

        async def get_sec():
            return await cache.sys.lpop("jobs.bars_validation.scope")

    else:

        async def get_sec():
            return secs.pop() if len(secs) else None

    errors = 0
    while code := await get_sec():
        try:
            for day in tf.day_frames[(tf.day_frames >= start) & (tf.day_frames <= end)]:
                expected = await get_checksum(day)
                if expected and expected.get(code):
                    actual = await calc_checksums(tf.int2date(day), [code])
                    d1 = actual.get(code)
                    d2 = expected.get(code)

                    missing1 = d2.keys() - d1  # local has no checksum
                    missing2 = d1.keys() - d2  # remote has no checksum
                    mismatch = {k for k in d1.keys() & d2 if d1[k] != d2[k]}

                    for k in missing1:
                        info = (
                            ValidationError.LOCAL_MISS,
                            day,
                            code,
                            k,
                            d1.get(k),
                            d2.get(k),
                        )
                        report.info("%s,%s,%s,%s,%s,%s", *info)
                        await emit.emit(Events.OMEGA_VALIDATION_ERROR, info)
                    for k in missing2:
                        info = (
                            ValidationError.REMOTE_MISS,
                            day,
                            code,
                            k,
                            d1.get(k),
                            d2.get(k),
                        )
                        report.info("%s,%s,%s,%s,%s,%s", *info)
                        await emit.emit(Events.OMEGA_VALIDATION_ERROR, info)
                    for k in mismatch:
                        info = (
                            ValidationError.MISMATCH,
                            day,
                            code,
                            k,
                            d1.get(k),
                            d2.get(k),
                        )
                        report.info("%s,%s,%s,%s,%s,%s", *info)
                        await emit.emit(Events.OMEGA_VALIDATION_ERROR, info)

                else:
                    logger.error("checksum for %s not found.", day)
                    info = (ValidationError.NO_CHECKSUM, day, None, None, None, None)
                    report.info("%s,%s,%s,%s,%s,%s", *info)
                    await emit.emit(Events.OMEGA_VALIDATION_ERROR, info)
        except Exception as e:
            logger.exception(e)
            errors += 1

    await emit.emit(Events.OMEGA_VALIDATION_ERROR, (ValidationError.UNKNOWN, errors))
    logger.warning("do_validation meet %s unknown errors", errors)

async def on_validation_error(report: tuple):
    """
    Args:
        report: object like ::(reason, day, code, frame, local, remote)

    Returns:

    """
    global validation_errors, no_validation_error_days

    # todo: raise no checksum issue
    if report[0] == ValidationError.UNKNOWN:
        no_validation_error_days = set()
    else:
        validation_errors.append(report)
        if report[1] is not None:
            no_validation_error_days -= {report[1]}

async def start_validation():
    """
    将待校验的证券按CPU个数均匀划分，创建与CPU个数相同的子进程来执行校验。校验的起始时间由数据
    库中jobs.bars_validation.range.start和jobs.bars_validation.range.stop来决定，每次校验
    结束后，将jobs.bars_validation.range.start更新为校验截止的最后交易日。如果各个子进程报告
    的截止交易日不一样（比如发生了异常），则使用最小的交易日。
    """
    global validation_errors, no_validation_error_days
    validation_errors = []

    secs = Securities()

    cpu_count = psutil.cpu_count()

    # to check if the range is right
    pl = cache.sys.pipeline()
    pl.get("jobs.bars_validation.range.start")
    pl.get("jobs.bars_validation.range.end")
    start, end = await pl.execute()

    if start is None:
        if cfg.omega.validation.start is None:
            logger.warning("start of validation is not specified, validation aborted.")
            return
        else:
            start = tf.date2int(arrow.get(cfg.omega.validation.start))
    else:
        start = int(start)

    if end is None:
        end = tf.date2int(tf.floor(arrow.now().date(), FrameType.DAY))
    else:
        end = int(end)

    assert start <= end

    no_validation_error_days = set(
        tf.day_frames[(tf.day_frames >= start) & (tf.day_frames <= end)]
    )

    # fixme: do validation per frame_type
    # fixme: test fail. Rewrite this before 0.6 releases
    codes = secs.choose(cfg.omega.sync)
    await cache.sys.delete("jobs.bars_validation.scope")
    await cache.sys.lpush("jobs.bars_validation.scope", *codes)

    logger.info("start validation %s secs from %s to %s.", len(codes), start, end)
    emit.register(Events.OMEGA_VALIDATION_ERROR, on_validation_error)

    t0 = time.time()

    code = (
        "from omega.core.sanity import do_validation_process_entry; "
        "do_validation_process_entry()"
    )

    procs = []
    for i in range(cpu_count):
        proc = subprocess.Popen([sys.executable, "-c", code], env=os.environ)
        procs.append(proc)

    timeout = 3600
    while timeout > 0:
        await asyncio.sleep(2)
        timeout -= 2
        for proc in procs:
            proc.poll()

        if all([proc.returncode is not None for proc in procs]):
            break

    if timeout <= 0:
        for proc in procs:
            try:
                os.kill(proc.pid, signal.SIGTERM)
            except Exception:
                pass

    # set next start point
    validation_days = set(
        tf.day_frames[(tf.day_frames >= start) & (tf.day_frames <= end)]
    )
    diff = validation_days - no_validation_error_days
    if len(diff):
        last_no_error_day = min(diff)
    else:
        last_no_error_day = end

    await cache.sys.set("jobs.bars_validation.range.start", last_no_error_day)
    elapsed = time.time() - t0
    logger.info(
        "Validation cost %s seconds, validation will start at %s next time",
        elapsed,
        last_no_error_day,
    )

async def adjust_range(batch: int = 500):
    """adjust secs's range after archive bars imported"""
    cur = b"0"
    key = "archive.ranges.*"
    logger.info("start adjust range")
    while cur:
        cur, keys = await cache.sys.scan(cur, match=key, count=batch)
        if not keys:
            continue

        pl = cache.security.pipeline()
        for item in keys:
            try:
                values = [int(v) for v in await cache.sys.lrange(item, 0, -1)]
                values.sort()

                arc_head, arc_tail = values[0], values[-1]

                code_frame_key = item.replace("archive.ranges.", "")
                head, tail = await cache.security.hmget(code_frame_key, "head", "tail")

                head = int(head) if head is not None else None
                tail = int(tail) if tail is not None else None

                # head, tail, arc_head, arc_tail should be all frame-aligned
                if head is None or tail is None:
                    head, tail = arc_head, arc_tail
                elif arc_tail < head or arc_head > tail:
                    head, tail = arc_head, arc_tail
                else:
                    head = min(arc_head, head)
                    tail = max(arc_tail, tail)
                pl.hset(code_frame_key, "head", head)
                pl.hset(code_frame_key, "tail", tail)
            except Exception as e:
                logger.exception(e)
                logger.warning("failed to set range for %s", code_frame_key)

        await pl.execute()

async def clear_range():
    """clear cached secs's range before/after import archive bars"""
    key = "archive.ranges.*"
    keys = await cache.sys.keys(key)

    if keys:
        await cache.sys.delete(*keys)

def main(months: str, cats: str, archive_server: str = None):
    """允许将本模块以独立进程运行，以支持多进程

    Args:
        months (str): 逗号分隔的月列表。格式如202012
        cats (str): 逗号分隔的类别列表，如"stock,index"
    """
    config_dir = get_config_dir()
    cfg = cfg4py.init(config_dir, False)

    if archive_server:
        cfg.omega.urls.archive = archive_server

    months = str(months)
    months = [int(x) for x in months.split(",") if x]
    cats = [x for x in cats.split(",")]

    asyncio.run(_main(months, cats))

def load_sync_params(frame_type: FrameType) -> dict:
    """根据指定的frame_type，从配置文件中加载同步参数

    Args:
        frame_type (FrameType): [description]

    Returns:
        dict: see @[omega.jobs.syncjobs.parse_sync_params]
    """
    for item in cfg.omega.sync.bars:
        if item.get("frame") == frame_type.value:
            try:
                secs, frame_type, start, stop, delay = parse_sync_params(**item)
                return item
            except Exception as e:
                logger.exception(e)
                logger.warning("failed to parse %s", item)
                return None

    return None

def parse_sync_params(
    frame: Union[str, Frame],
    cat: List[str] = None,
    start: Union[str, datetime.date] = None,
    stop: Union[str, Frame] = None,
    delay: int = 0,
    include: str = "",
    exclude: str = "",
) -> Tuple:
    """按照[使用手册](usage.md#22-如何同步K线数据)中的规则，解析和补全同步参数。

    如果`frame_type`为分钟级，则当`start`指定为`date`类型时，自动更正为对应交易日的起始帧；
    当`stop`为`date`类型时，自动更正为对应交易日的最后一帧。
    Args:
        frame (Union[str, Frame]): frame type to be sync.  The word ``frame`` is used
            here for easy understand by end user. It actually implies "FrameType".
        cat (List[str]): which catetories is about to be synced. Should be one of
            ['stock', 'index']. Defaults to None.
        start (Union[str, datetime.date], optional): [description]. Defaults to None.
        stop (Union[str, Frame], optional): [description]. Defaults to None.
        delay (int, optional): [description]. Defaults to 5.
        include (str, optional): which securities should be included, seperated by
            space, for example, "000001.XSHE 000004.XSHE". Defaults to empty string.
        exclude (str, optional):  which securities should be excluded, seperated by
            a space. Defaults to empty string.

    Returns:
        - codes (List[str]): 待同步证券列表
        - frame_type (FrameType):
        - start (Frame):
        - stop (Frame):
        - delay (int):
    """
    frame_type = FrameType(frame)

    if frame_type in tf.minute_level_frames:
        if stop:
            stop = arrow.get(stop, tzinfo=cfg.tz)
            if stop.hour == 0:  # 未指定有效的时间帧，使用当日结束帧
                stop = tf.last_min_frame(tf.day_shift(stop.date(), 0), frame_type)
            else:
                stop = tf.floor(stop, frame_type)
        else:
            stop = tf.floor(arrow.now(tz=cfg.tz).datetime, frame_type)

        if stop > arrow.now(tz=cfg.tz):
            raise ValueError(f"请勿将同步截止时间设置在未来: {stop}")

        if start:
            start = arrow.get(start, tzinfo=cfg.tz)
            if start.hour == 0:  # 未指定有效的交易帧，使用当日的起始帧
                start = tf.first_min_frame(tf.day_shift(start.date(), 0), frame_type)
            else:
                start = tf.floor(start, frame_type)
        else:
            start = tf.shift(stop, -999, frame_type)
    else:
        stop = (stop and arrow.get(stop).date()) or arrow.now().date()
        if stop == arrow.now().date():
            stop = arrow.now(tz=cfg.tz)

        stop = tf.floor(stop, frame_type)
        start = tf.floor((start and arrow.get(start).date()), frame_type) or tf.shift(
            stop, -1000, frame_type
        )

    secs = Securities()
    codes = secs.choose(cat or [])

    exclude = map(lambda x: x, exclude.split(" "))
    codes = list(set(codes) - set(exclude))

    include = list(filter(lambda x: x, include.split(" ")))
    codes.extend(include)

    return codes, frame_type, start, stop, int(delay)

async def sync_bars(params: dict):
    """sync bars on signal OMEGA_DO_SYNC received

    Args:
        params (dict): composed of the following:
            ```
            {
                secs (List[str]): 待同步的证券标的.如果为None或者为空，则从数据库中轮询
                frame_type (FrameType):k线的帧类型
                start (Frame): k线起始时间
                stop (Frame): k线结束时间
            }
            ```
    Returns:
        [type]: [description]
    """
    secs, frame_type, start, stop = (
        params.get("secs"),
        params.get("frame_type"),
        params.get("start"),
        params.get("stop"),
    )

    if secs is not None:
        logger.info(
            "sync bars with %s(%s ~ %s) for given %s secs",
            frame_type,
            start,
            stop,
            len(secs),
        )

        async def get_sec():
            return secs.pop() if len(secs) else None

    else:
        logger.info(
            "sync bars with %s(%s ~ %s) in polling mode", frame_type, start, stop
        )

        async def get_sec():
            return await cache.sys.lpop(key_scope)

    key_scope = f"jobs.bars_sync.scope.{frame_type.value}"

    if start is None or frame_type is None:
        raise ValueError("you must specify a start date/frame_type for sync")

    if stop is None:
        stop = tf.floor(arrow.now(tz=cfg.tz), frame_type)

    while code := await get_sec():
        try:
            await sync_bars_for_security(code, frame_type, start, stop)
        except FetcherQuotaError as e:
            logger.warning("Quota exceeded when syncing %s. Sync aborted.", code)
            logger.exception(e)
            return  # stop the sync
        except Exception as e:
            logger.warning("Failed to sync %s", code)
            logger.exception(e)

    elapsed = await _stop_job_timer("sync")
    logger.info("%s finished quotes sync in %s seconds", os.getpid(), elapsed)

async def sync_calendar():
    """从上游服务器获取所有交易日，并计算出周线帧和月线帧

    Returns:
    """
    trade_days = await aq.get_all_trade_days()
    if trade_days is None or len(trade_days) == 0:
        logger.warning("failed to fetch trade days.")
        return None

    tf.day_frames = [tf.date2int(x) for x in trade_days]
    weeks = []
    last = trade_days[0]
    for cur in trade_days:
        if cur.weekday() < last.weekday() or (cur - last).days >= 7:
            weeks.append(last)
        last = cur

    if weeks[-1] < last:
        weeks.append(last)

    tf.week_frames = [tf.date2int(x) for x in weeks]
    await cache.save_calendar("week_frames", map(tf.date2int, weeks))

    months = []
    last = trade_days[0]
    for cur in trade_days:
        if cur.day < last.day:
            months.append(last)
        last = cur
    months.append(last)

    tf.month_frames = [tf.date2int(x) for x in months]
    await cache.save_calendar("month_frames", map(tf.date2int, months))
    logger.info("trade_days is updated to %s", trade_days[-1])

async def sync_security_list():
    """更新证券列表

    注意证券列表在AbstractQuotesServer取得时就已保存，此处只是触发
    """
    secs = await aq.get_security_list()
    logger.info("%s secs are fetched and saved.", len(secs))

async def trigger_bars_sync(sync_params: dict = None, force=False):
    """初始化bars_sync的任务，发信号给各quotes_fetcher进程以启动同步。

    Args:
        frame_type (FrameType): 要同步的帧类型
        sync_params (dict): 同步参数
            ```
            {
                start: 起始帧
                stop: 截止帧
                frame: 帧类型
                delay: 延迟启动时间，以秒为单位
                cat: 证券分类，如stock, index等
                delay: seconds for sync to wait.
            }
            ```
            see more @[omega.jobs.syncjobs.parse_sync_params][]
        force: 即使当前不是交易日，是否也强行进行同步。
    Returns:

    """
    if not force and not tf.is_trade_day(arrow.now()):
        return

    codes, frame_type, start, stop, delay = parse_sync_params(**sync_params)
    key_scope = f"jobs.bars_sync.scope.{frame_type.value}"

    if len(codes) == 0:
        logger.warning("no securities are specified for sync %s", frame_type)
        return

    fmt_str = "sync from %s to %s in frame_type(%s) for %s secs"
    logger.info(fmt_str, start, stop, frame_type, len(codes))

    # secs are stored into cache, so each fetcher can polling it
    pl = cache.sys.pipeline()
    pl.delete(key_scope)
    pl.lpush(key_scope, *codes)
    await pl.execute()

    await asyncio.sleep(delay)
    await _start_job_timer("sync")

    await emit.emit(
        Events.OMEGA_DO_SYNC, {"frame_type": frame_type, "start": start, "stop": stop}
    )

    fmt_str = "send trigger sync event to fetchers: from %s to %s in frame_type(%s) for %s secs"
    logger.info(fmt_str, start, stop, frame_type, len(codes))

async def trigger_single_worker_sync(_type: str, params: dict = None):
    """启动只需要单个quotes fetcher进程来完成的数据同步任务

    比如交易日历、证券列表等如果需要同时启动多个quotes fetcher进程来完成数据同步任务，应该通过
    pyemit来发送广播消息。

    Args:
        _type: the type of data to be synced, either ``calendar`` or ``ecurity_list``
    """
    url = cfg.omega.urls.quotes_server
    if _type == "calendar":
        url += "/jobs/sync_calendar"
    elif _type == "security_list":
        url += "/jobs/sync_security_list"
    else:
        raise ValueError(f"{_type} is not supported sync type.")

    async with aiohttp.ClientSession() as client:
        try:
            async with client.post(url, data=params) as resp:
                if resp.status != 200:
                    logger.warning("failed to trigger %s sync", _type)
                else:
                    return await resp.json()
        except Exception as e:
            logger.exception(e)