"""同一时点跨股票排名;上市历史、池生效和输入重排分别验收。""" import itertools import json from math import isfinite prices = {'A': [10, 11, 12], 'B': [10, 12, 11], 'C': [None, 20, 22]} active_from = {'A': 0, 'B': 0, 'C': 1} def select(data, membership): length = len(next(iter(data.values()))) if any(len(values) != length for values in data.values()): raise ValueError('必须先显式对齐共同交易期') result = [] for i in range(length): scores = {} for symbol, values in data.items(): if i == 0 or i < membership[symbol]: continue past, current = values[i - 1:i + 1] if (past is not None and current is not None and isfinite(past) and isfinite(current) and past > 0 and current > 0): scores[symbol] = current / past - 1 chosen = sorted(scores, key=lambda s: (-scores[s], s))[:1] result.append({'i': i, 'scores': scores, 'chosen': chosen, 'signals': {s: int(s in chosen) for s in data}}) return result expected = select(prices, active_from) assert [r['chosen'] for r in expected] == [[], ['B'], ['C']] assert 'C' not in expected[1]['scores'] for order in itertools.permutations(prices): assert select({s: prices[s] for s in order}, active_from) == expected late = select(prices, dict(active_from, C=3)) assert late[2]['chosen'] == ['A'] naive = [s for s in prices if prices[s][0] and prices[s][1] > prices[s][0]] assert naive == ['A', 'B'] and naive != expected[1]['chosen'] assert select({s: xs + [1000] for s, xs in prices.items()}, active_from)[:3] == expected try: select({'A': [10, 11], 'B': [10]}, {'A': 0, 'B': 0}) except ValueError as error: assert '对齐' in str(error) else: raise AssertionError('未对齐长度必须明确拒绝') print(json.dumps({'cross_sections': expected, 'late_membership_last': late[2], 'permutations_checked': 6, 'per_stock_positive_counterexample': naive}))