> ## Documentation Index > Fetch the complete documentation index at: https://docs.clypt.ai/llms.txt > Use this file to discover all available pages before exploring further. # Alpha 101 > 101 Formulaic Alphas — systematic alpha generation from Kakushadze (2016) ## Overview Implementation of [101 Formulaic Alphas](https://arxiv.org/abs/1601.00991) (Kakushadze, Z., 2016). **101 alphas** available, each as a standalone operator inheriting from [`AlphaOperator`](/operators/signals/alpha-operator). All Alpha 101 operators use the **AlphaOperator DSL** — numba-accelerated helpers for time-series and cross-sectional operations. Each alpha's `compute_signal()` is typically 5-15 lines of DSL calls. **Lookback validation**: Each alpha validates that input lookback is sufficient for its window parameters at instantiation time. Insufficient lookback raises `ValueError` immediately instead of producing silent zeros. ### Input Categories | Category | Inputs | Example Alphas | | ---------------- | ----------------------------- | ---------------------- | | **Price-only** | close (or open+close) | A001, A004, A008, A033 | | **Price-volume** | close+volume (or open+volume) | A003, A007, A012, A013 | | **Multi-field** | high+low+close+volume | A011, A025, A041, A055 | | **All fields** | open+high+low+close+volume | A005, A036, A062, A071 | See the [Alpha101 IC Analysis notebook](https://github.com/Clypt/clyptq/blob/master/examples/community/12_alpha101_comprehensive_backtest.ipynb) for comprehensive crypto backtesting results. ## Alpha Catalog | Alpha | Description | Key Parameters | | ----------------- | ---------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------- | | **Alpha101\_001** | Alpha #001: Volatility-based ranking signal. | `std_window=20`, `argmax_window=5` | | **Alpha101\_002** | Alpha #002: Volume-price correlation signal. | `delta_window=2`, `corr_window=6` | | **Alpha101\_003** | Alpha #003: Open-volume correlation signal. | `corr_window=10` | | **Alpha101\_004** | Alpha #004: Low price time-series rank signal. | `ts_rank_window=9` | | **Alpha101\_005** | Alpha #005: VWAP deviation signal. | `vwap_window=10` | | **Alpha101\_006** | Alpha #006: Open-volume correlation signal. | `corr_window=10` | | **Alpha101\_007** | Alpha #007: Volume-conditional price change signal. | `amount_window=20`, `delta_window=7`, `rank_window=60` | | **Alpha101\_008** | Alpha #008: Open-returns delayed comparison signal. | `sum_window=5`, `delay_window=10` | | **Alpha101\_009** | Alpha #009: Price change direction consistency signal. | `delta_window=1`, `consistency_window=5` | | **Alpha101\_010** | Alpha #010: Price change direction consistency ranking. | `delta_window=1`, `consistency_window=4` | | **Alpha101\_011** | Alpha #011: VWAP-close deviation and volume change signal. | `window=3` | | **Alpha101\_012** | Alpha #012: Volume-price divergence signal. | `delta_window=1` | | **Alpha101\_013** | Alpha #013: Close-volume covariance signal. | `cov_window=5` | | **Alpha101\_014** | Alpha #014: Returns delta and open-volume correlation signal. | `delta_window=3`, `corr_window=10` | | **Alpha101\_015** | Alpha #015: High-volume correlation ranking signal. | `corr_window=3`, `sum_window=3` | | **Alpha101\_016** | Alpha #016: High-volume covariance ranking signal. | `cov_window=5` | | **Alpha101\_017** | Alpha #017: Complex close-volume momentum signal. | `ts_rank_window1=10`, `ts_rank_window2=5`, `amount_mean_window=20` | | **Alpha101\_018** | Alpha #018: Close-open volatility and correlation signal. | `std_window=5`, `corr_window=10` | | **Alpha101\_019** | Alpha #019: Price direction and long-term returns signal. | `delay_window=7`, `delta_window=7`, `returns_sum_window=250` | | **Alpha101\_020** | Alpha #020: Opening gap ranking signal. | `delay_window=1` | | **Alpha101\_021** | Alpha #021: Close volatility and volume ratio signal. | `short_window=2`, `long_window=8`, `amount_window=20` | | **Alpha101\_022** | Alpha #022: High-volume correlation change signal. | `corr_window=5`, `delta_window=5`, `std_window=20` | | **Alpha101\_023** | Alpha #023: High price breakout signal. | `avg_window=20`, `delta_window=2` | | **Alpha101\_024** | Alpha #024: Long-term average change rate signal. | `long_window=100`, `short_delta=3`, `threshold=0.05` | | **Alpha101\_025** | Alpha #025: Returns-amount-VWAP composite signal. | `amount_window=20` | | **Alpha101\_026** | Alpha #026: Volume-high time-series correlation signal. | `ts_rank_window=5`, `corr_window=5`, `max_window=3` | | **Alpha101\_027** | Alpha #027: Volume-VWAP correlation ranking signal. | `corr_window=6`, `sum_window=2` | | **Alpha101\_028** | Alpha #028: Amount-low correlation with mid-price signal. | `amount_window=20`, `corr_window=5` | | **Alpha101\_029** | Alpha #029: Complex nested ranking signal. | `delta_window=5`, `delay_window=6`, `ts_rank_window=5` | | **Alpha101\_030** | Alpha #030: Price direction pattern with volume ratio signal. | `short_volume_window=5`, `long_volume_window=20` | | **Alpha101\_031** | Alpha #031: Multi-ranking decay + amount-low correlation. | `long_delta=10`, `short_delta=3`, `decay_window=10` | | **Alpha101\_032** | Alpha #032: Mean reversion with VWAP correlation signal. | `mean_window=7`, `delay_window=5`, `corr_window=230` | | **Alpha101\_033** | Alpha #033: Open-close ratio momentum signal. | — | | **Alpha101\_034** | Alpha #034: Volatility ratio and price change signal. | `short_std=2`, `long_std=5`, `delta_window=1` | | **Alpha101\_035** | Alpha #035: Volume-price-returns time-series ranking signal. | `volume_window=32`, `price_window=16`, `returns_window=32` | | **Alpha101\_036** | Alpha #036: Weighted multi-factor composite signal. | — | | **Alpha101\_037** | Alpha #037: Long-term open-close correlation signal. | `corr_window=200`, `delay_window=1` | | **Alpha101\_038** | Alpha #038: Close time-series rank with close/open ratio signal. | `ts_rank_window=10` | | **Alpha101\_039** | Alpha #039: Price delta with decayed volume ratio signal. | `delta_window=7`, `amount_window=20`, `decay_window=9` | | **Alpha101\_040** | Alpha #040: High volatility with high-volume correlation signal. | `std_window=10`, `corr_window=10` | | **Alpha101\_041** | Alpha #041: Geometric mean minus VWAP signal. | — | | **Alpha101\_042** | Alpha #042: VWAP-close difference to sum ratio signal. | — | | **Alpha101\_043** | Alpha #043: Volume ratio and price delta time-series ranking signal. | `amount_window=20`, `volume_rank_window=20`, `delta_window=7` | | **Alpha101\_044** | Alpha #044: High-volume rank correlation signal. | `corr_window=5` | | **Alpha101\_045** | Alpha #045: Delayed close mean with correlations signal. | `delay_window=5`, `sum_window=20`, `short_sum=5` | | **Alpha101\_046** | Alpha #046: Multi-period slope comparison signal. | — | | **Alpha101\_047** | Alpha #047: Complex price-volume-VWAP signal. | `amount_window=20`, `high_window=5`, `vwap_delay=5` | | **Alpha101\_048** | Alpha #048: Price change correlation with volatility signal. | `corr_window=250`, `vol_window=250` | | **Alpha101\_049** | Alpha #049: Slope comparison with threshold signal. | `threshold=-0.1` | | **Alpha101\_050** | Alpha #050: Volume-VWAP correlation max signal. | `corr_window=5`, `max_window=5` | | **Alpha101\_051** | Alpha #051: Slope comparison with threshold signal. | `threshold=-0.05` | | **Alpha101\_052** | Alpha #052: Low minimum change with returns and volume signal. | `low_window=5`, `returns_long=240`, `returns_short=20` | | **Alpha101\_053** | Alpha #053: Price position delta signal. | `delta_window=9` | | **Alpha101\_054** | Alpha #054: Price ratio with power signal. | `power=5` | | **Alpha101\_055** | Alpha #055: Stochastic-volume correlation signal. | `stoch_window=12`, `corr_window=6` | | **Alpha101\_056** | Alpha #056: Returns ratio and cap product signal. | `returns_window1=10`, `returns_window2=2`, `nested_window=3` | | **Alpha101\_057** | Alpha #057: Close-VWAP with argmax decay signal. | `argmax_window=30`, `decay_window=2` | | **Alpha101\_058** | Alpha #058: Demeaned VWAP-volume correlation decay rank signal. | `corr_window=4`, `decay_window=8`, `rank_window=6` | | **Alpha101\_059** | Alpha #059: Weighted VWAP-volume correlation decay rank signal. | `corr_window=4`, `decay_window=16`, `rank_window=8` | | **Alpha101\_060** | Alpha #060: Price position volume vs argmax signal. | `argmax_window=10` | | **Alpha101\_061** | Alpha #061: VWAP range vs amount correlation rank signal. | `vwap_min_window=16`, `amount_window=180`, `corr_window=18` | | **Alpha101\_062** | Alpha #062: VWAP-amount correlation vs price rank signal. | `amount_window=20`, `sum_window=22`, `corr_window=10` | | **Alpha101\_063** | Alpha #063: Demeaned close delta vs weighted price-amount correlation signal. | `delta_window=2`, `decay_window1=8`, `vwap_weight=0.318108` | | **Alpha101\_064** | Alpha #064: Weighted open-low amount correlation vs mid-VWAP delta signal. | `weight=0.178404`, `sum_window=13`, `amount_window=120` | | **Alpha101\_065** | Alpha #065: Weighted open-VWAP amount correlation vs open range signal. | `weight=0.00817205`, `amount_window=60`, `sum_window=9` | | **Alpha101\_066** | Alpha #066: VWAP delta decay rank plus low-VWAP ratio ts\_rank signal. | `delta_window=4`, `decay_window1=7`, `decay_window2=11` | | **Alpha101\_067** | Alpha #067: High range rank power by demeaned VWAP-amount correlation rank signal. | `high_min_window=2`, `amount_window=20`, `corr_window=6` | | **Alpha101\_068** | Alpha #068: High-amount correlation ts\_rank vs weighted close-low delta signal. | `amount_window=15`, `corr_window=9`, `rank_window=14` | | **Alpha101\_069** | Alpha #69: Demeaned VWAP delta max rank power by weighted price-amount correlation ts\_rank. | `delta_window=3`, `max_window=5`, `weight=0.490655` | | **Alpha101\_070** | Alpha #70: VWAP delta rank power by demeaned close-amount correlation ts\_rank. | `delta_window=1`, `amount_window=50`, `corr_window=18` | | **Alpha101\_071** | Alpha #71: Close-amount correlation decay rank vs price difference decay rank max. | `close_rank_window=3`, `amount_window=180`, `amount_rank_window=12` | | **Alpha101\_072** | Alpha #72: Mid price-amount correlation decay rank ratio. | `amount_window=40`, `corr_window1=9`, `decay_window1=10` | | **Alpha101\_073** | Alpha #73: VWAP delta decay rank vs weighted price change rate decay rank max. | `vwap_delta_window=5`, `decay_window1=3`, `weight=0.147155` | | **Alpha101\_074** | Alpha #74: Close-amount correlation rank vs weighted high-VWAP volume correlation rank. | `amount_window=30`, `sum_window=37`, `corr_window1=15` | | **Alpha101\_075** | Alpha #75: VWAP-volume correlation rank vs low-amount rank correlation rank. | `corr_window1=4`, `amount_window=50`, `corr_window2=12` | | **Alpha101\_076** | Alpha #76: VWAP delta decay rank vs demeaned low-amount correlation decay rank max. | `delta_window=1`, `decay_window1=12`, `amount_window=81` | | **Alpha101\_077** | Alpha #77: Price difference decay rank vs mid-amount correlation decay rank min. | `decay_window1=20`, `amount_window=40`, `corr_window=3` | | **Alpha101\_078** | Alpha #78: Weighted low-VWAP amount correlation rank power by VWAP-volume rank correlation rank. | `weight=0.352233`, `sum_window=20`, `amount_window=40` | | **Alpha101\_079** | Alpha #79: Demeaned weighted close-open delta rank vs VWAP-amount ts\_rank correlation rank. | `weight=0.60733`, `delta_window=1`, `vwap_rank_window=4` | | **Alpha101\_080** | Alpha #80: Demeaned weighted open-high delta sign rank power by high-amount correlation ts\_rank. | `weight=0.868128`, `delta_window=4`, `amount_window=10` | | **Alpha101\_081** | Alpha #81: VWAP-amount correlation product log rank vs VWAP-volume rank correlation rank. | `amount_window=10`, `sum_window=50`, `corr_window1=8` | | **Alpha101\_082** | Alpha #82: Open delta decay rank vs demeaned volume-open correlation decay rank min. | `delta_window=1`, `decay_window1=15`, `corr_window=17` | | **Alpha101\_083** | Alpha #83: Delayed range ratio rank times double volume rank ratio. | `mean_window=5`, `delay_period=2` | | **Alpha101\_084** | Alpha #84: VWAP max difference ts\_rank power by close delta. | `max_window=15`, `rank_window=21`, `delta_window=5` | | **Alpha101\_085** | Alpha #85: Weighted high-close amount correlation rank power by mid-volume ts\_rank correlation rank. | `weight=0.876703`, `amount_window=30`, `corr_window1=10` | | **Alpha101\_086** | Alpha #86: Close-amount correlation ts\_rank vs price sum difference rank. | `amount_window=20`, `sum_window=15`, `corr_window=6` | | **Alpha101\_087** | Alpha #87: Weighted close-VWAP delta decay rank vs demeaned amount-close correlation abs decay rank max. | `weight=0.369701`, `delta_window=2`, `decay_window1=3` | | **Alpha101\_088** | Alpha #88: Price rank sum difference decay rank vs close-amount ts\_rank correlation decay rank min. | `decay_window1=8`, `close_rank_window=8`, `amount_window=60` | | **Alpha101\_089** | Alpha #89: Weighted low-amount correlation decay rank minus demeaned VWAP delta decay rank. | `amount_window=10`, `corr_window=7`, `decay_window1=6` | | **Alpha101\_090** | Alpha #90: Close max difference rank power by demeaned amount-low correlation ts\_rank. | `max_window=5`, `amount_window=40`, `corr_window=5` | | **Alpha101\_091** | Alpha #91: Double decayed close-volume correlation ts\_rank minus VWAP-amount correlation decay rank. | `corr_window1=10`, `decay_window1=16`, `decay_window2=4` | | **Alpha101\_092** | Alpha #92: Mid-close vs low-open comparison decay rank min with low-amount rank correlation decay rank. | `decay_window1=15`, `rank_window1=19`, `amount_window=30` | | **Alpha101\_093** | Alpha #93: Demeaned VWAP-amount correlation decay ts\_rank divided by weighted close-VWAP delta decay rank. | `amount_window=81`, `corr_window=17`, `decay_window1=20` | | **Alpha101\_094** | Alpha #94: VWAP-min VWAP difference rank power by VWAP-amount ts\_rank correlation ts\_rank. | `min_window=12`, `vwap_rank_window=20`, `amount_window=60` | | **Alpha101\_095** | Alpha #95: Open-min open difference rank less than mid-amount correlation rank power ts\_rank. | `min_window=12`, `sum_window=19`, `amount_window=40` | | **Alpha101\_096** | Alpha #96: VWAP-volume rank correlation decay ts\_rank vs close-amount ts\_rank correlation argmax decay ts\_rank max. | `corr_window1=4`, `decay_window1=4`, `rank_window1=8` | | **Alpha101\_097** | Alpha #97: Demeaned weighted low-VWAP delta decay rank minus low-amount ts\_rank correlation decay ts\_rank. | `weight=0.721001`, `delta_window=3`, `decay_window1=20` | | **Alpha101\_098** | Alpha #98: VWAP-amount correlation decay rank minus open-amount rank correlation argmin decay rank. | `amount_window1=5`, `sum_window=26`, `corr_window1=5` | | **Alpha101\_099** | Alpha #99: Mid-amount sum correlation rank less than low-volume correlation rank comparison. | `sum_window=20`, `amount_window=60`, `corr_window1=9` | | **Alpha101\_100** | Alpha #100: Complex multi-demeaned price position-volume and correlation factor. | `amount_window=20`, `corr_window=5`, `argmin_window=30` | | **Alpha101\_101** | Alpha #101: Price change divided by price range. | `epsilon=0.001` | ## Usage Pattern All Alpha 101 operators follow the same pattern: ```python theme={null} from clyptq.apps.trading.operators.signal.alpha.alpha_101 import Alpha101_001 graph.add_node("alpha_001", Alpha101_001( Input("FIELD:binance:futures:ohlcv:close", timeframe="1m", lookback=20) )) ``` ## Source Code Full `compute()` implementations — no hidden logic. Alpha #001: Volatility-based ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] values = data.value last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(values.shape) > 1 and len(values) >= 2: # Calculate returns with np.errstate(divide='ignore', invalid='ignore'): returns = np.diff(values, axis=0) / values[:-1] if len(returns) >= self._std_window: # Rolling std of returns returns_std = np.std(returns[-self._std_window:], axis=0) # Condition: returns < 0 last_returns = returns[-1] returns_negative = last_returns < 0 # condition(returns_negative, returns_std, close) condition_result = np.where(returns_negative, returns_std, values[-1]) # pow(condition_result, 2) powered = condition_result ** 2 # ts_argmax over argmax_window if len(values) >= self._argmax_window: window_data = np.broadcast_to( powered, (self._argmax_window, n_symbols) ) argmax = np.argmax(window_data, axis=0).astype(float) else: argmax = np.zeros(n_symbols) # rank (cross-sectional) compute_mask = exists & valid ranked = np.zeros(n_symbols) if compute_mask.any(): valid_vals = argmax[compute_mask] ranks = (np.argsort(np.argsort(valid_vals)) + 1) / len(valid_vals) ranked[compute_mask] = ranks # sub(ranked, 0.5) alpha = ranked - 0.5 result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #002: Volume-price correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, open_data, volume_data = data[0], data[1], data[2] else: close_data = open_data = volume_data = data close = close_data.value open_ = open_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._corr_window + self._delta_window: # log(volume) with np.errstate(divide='ignore', invalid='ignore'): log_volume = np.log(volume) # ts_delta(log_volume, delta_window) if len(log_volume) > self._delta_window: volume_delta = log_volume[self._delta_window:] - log_volume[:-self._delta_window] else: volume_delta = np.zeros_like(log_volume) # price_change = (close - open) / open with np.errstate(divide='ignore', invalid='ignore'): price_change = (close - open_) / open_ # Use last corr_window for correlation if len(volume_delta) >= self._corr_window and len(price_change) >= self._corr_window: vol_window = volume_delta[-self._corr_window:] price_window = price_change[-self._corr_window:] # Cross-sectional rank for each time step, then correlate compute_mask = exists & valid for i in range(n_symbols): if compute_mask[i]: vol_series = vol_window[:, i] price_series = price_window[:, i] # Filter out NaN/Inf valid_mask = ~(np.isnan(vol_series) | np.isnan(price_series) | np.isinf(vol_series) | np.isinf(price_series)) if valid_mask.sum() >= 3: vol_valid = vol_series[valid_mask] price_valid = price_series[valid_mask] # Correlation vol_mean = np.mean(vol_valid) price_mean = np.mean(price_valid) vol_std = np.std(vol_valid) price_std = np.std(price_valid) if vol_std > 0 and price_std > 0: corr = np.mean((vol_valid - vol_mean) * (price_valid - price_mean)) / (vol_std * price_std) result[i] = -corr # Negate correlation result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #003: Open-volume correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, volume_data = data[0], data[1] else: open_data = volume_data = data open_ = open_data.value volume = volume_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(open_.shape) > 1 and len(open_) >= self._corr_window: compute_mask = exists & valid for i in range(n_symbols): if compute_mask[i]: open_series = open_[-self._corr_window:, i] vol_series = volume[-self._corr_window:, i] valid_mask = ~(np.isnan(open_series) | np.isnan(vol_series) | np.isinf(open_series) | np.isinf(vol_series)) if valid_mask.sum() >= 3: open_valid = open_series[valid_mask] vol_valid = vol_series[valid_mask] open_std = np.std(open_valid) vol_std = np.std(vol_valid) if open_std > 0 and vol_std > 0: corr = np.corrcoef(open_valid, vol_valid)[0, 1] result[i] = -corr result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #004: Low price time-series rank signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] values = data.value # low prices last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(values.shape) > 1 and len(values) >= self._ts_rank_window: compute_mask = exists & valid window_data = values[-self._ts_rank_window:] for i in range(n_symbols): if compute_mask[i]: series = window_data[:, i] valid_mask = ~np.isnan(series) if valid_mask.sum() >= 2: valid_series = series[valid_mask] current_val = valid_series[-1] # Time-series rank: position of current value ts_rank = (np.sum(valid_series <= current_val) / len(valid_series)) result[i] = -ts_rank # Inverted result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #005: VWAP deviation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data, volume_data = data else: open_data = high_data = low_data = close_data = volume_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._vwap_window: # Calculate VWAP: (high + low + close) / 3 * volume / sum(volume) typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # vwap_avg = ts_sum(vwap, window) / window vwap_window = vwap[-self._vwap_window:] vwap_avg = np.mean(vwap_window, axis=0) # open_vwap_diff = open - vwap_avg open_vwap_diff = open_[-1] - vwap_avg # close_vwap_diff = close - vwap (current) close_vwap_diff = close[-1] - vwap[-1] compute_mask = exists & valid # Cross-sectional rank if compute_mask.any(): # rank(open_vwap_diff) open_vwap_ranked = np.zeros(n_symbols) valid_open_diff = open_vwap_diff[compute_mask] open_ranks = (np.argsort(np.argsort(valid_open_diff)) + 1) / len(valid_open_diff) open_vwap_ranked[compute_mask] = open_ranks # rank(close_vwap_diff) close_vwap_ranked = np.zeros(n_symbols) valid_close_diff = close_vwap_diff[compute_mask] close_ranks = (np.argsort(np.argsort(valid_close_diff)) + 1) / len(valid_close_diff) close_vwap_ranked[compute_mask] = close_ranks # alpha = rank(open-vwap_avg) * (-abs(rank(close-vwap))) alpha = open_vwap_ranked * (-np.abs(close_vwap_ranked)) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #006: Open-volume correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, volume_data = data[0], data[1] else: open_data = volume_data = data open_ = open_data.value volume = volume_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(open_.shape) > 1 and len(open_) >= self._corr_window: compute_mask = exists & valid for i in range(n_symbols): if compute_mask[i]: open_series = open_[-self._corr_window:, i] vol_series = volume[-self._corr_window:, i] valid_mask = ~(np.isnan(open_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: open_valid = open_series[valid_mask] vol_valid = vol_series[valid_mask] if np.std(open_valid) > 0 and np.std(vol_valid) > 0: corr = np.corrcoef(open_valid, vol_valid)[0, 1] result[i] = -corr result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #007: Volume-conditional price change signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data[0], data[1] else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._amount_window, self._delta_window, self._rank_window) if len(close.shape) > 1 and len(close) >= min_len: # amount = volume * close amount = volume * close # amount_mean = ts_mean(amount, amount_window) amount_mean = np.mean(amount[-self._amount_window:], axis=0) # condition_check = amount_mean < volume (current) condition_check = amount_mean < volume[-1] # close_delta = ts_delta(close, delta_window) if len(close) > self._delta_window: close_delta = close[-1] - close[-(self._delta_window + 1)] else: close_delta = np.zeros(n_symbols) # abs_close_delta abs_close_delta = np.abs(close_delta) # ts_rank(abs_close_delta, rank_window) - time-series rank compute_mask = exists & valid ts_ranked = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: # Get historical abs deltas for ts_rank if len(close) > self._delta_window: deltas = close[self._delta_window:, i] - close[:-self._delta_window, i] abs_deltas = np.abs(deltas) window = abs_deltas[-min(self._rank_window, len(abs_deltas)):] current_val = abs_close_delta[i] ts_ranked[i] = np.sum(window <= current_val) / len(window) # neg_ts_rank = ts_ranked * -1 neg_ts_rank = -ts_ranked # sign_delta = sign(close_delta) sign_delta = np.sign(close_delta) # true_value = neg_ts_rank * sign_delta true_value = neg_ts_rank * sign_delta # alpha = condition(condition_check, true_value, -1) alpha = np.where(condition_check, true_value, -1.0) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #008: Open-returns delayed comparison signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, close_data = data[0], data[1] else: open_data = close_data = data open_ = open_data.value close = close_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._sum_window + self._delay_window if len(close.shape) > 1 and len(close) >= min_len: # returns = ts_returns(close) with np.errstate(divide='ignore', invalid='ignore'): returns = np.diff(close, axis=0) / close[:-1] # open_sum = ts_sum(open, sum_window) # returns_sum = ts_sum(returns, sum_window) if len(returns) >= self._sum_window: open_sum_current = np.sum(open_[-self._sum_window:], axis=0) returns_sum_current = np.sum(returns[-self._sum_window:], axis=0) # current_product = open_sum * returns_sum current_product = open_sum_current * returns_sum_current # delayed_product = delay(current_product, delay_window) if len(open_) >= self._sum_window + self._delay_window: delay_idx = -(self._sum_window + self._delay_window) open_sum_delayed = np.sum(open_[delay_idx:delay_idx + self._sum_window], axis=0) returns_sum_delayed = np.sum(returns[delay_idx:delay_idx + self._sum_window], axis=0) delayed_product = open_sum_delayed * returns_sum_delayed else: delayed_product = np.zeros(n_symbols) # diff = current_product - delayed_product diff = current_product - delayed_product # rank(diff) cross-sectionally compute_mask = exists & valid ranked = np.zeros(n_symbols) if compute_mask.any(): valid_diff = diff[compute_mask] ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) ranked[compute_mask] = ranks # alpha = -ranked alpha = -ranked result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #009: Price change direction consistency signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] values = data.value # close prices last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._delta_window + self._consistency_window if len(values.shape) > 1 and len(values) >= min_len: # close_delta = ts_delta(close, delta_window) if len(values) > self._delta_window: close_delta = values[self._delta_window:] - values[:-self._delta_window] else: close_delta = np.zeros_like(values) if len(close_delta) >= self._consistency_window: # min_delta = ts_min(close_delta, consistency_window) window_deltas = close_delta[-self._consistency_window:] min_delta = np.min(window_deltas, axis=0) # max_delta = ts_max(close_delta, consistency_window) max_delta = np.max(window_deltas, axis=0) # Current close_delta current_delta = close_delta[-1] # cond1 = min_delta > 0 (all positive) cond1 = min_delta > 0 # cond2 = max_delta < 0 (all negative) cond2 = max_delta < 0 # neg_close_delta = close_delta * -1 neg_close_delta = -current_delta # inner_condition = condition(cond2, close_delta, neg_close_delta) inner_condition = np.where(cond2, current_delta, neg_close_delta) # alpha = condition(cond1, close_delta, inner_condition) alpha = np.where(cond1, current_delta, inner_condition) compute_mask = exists & valid result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #010: Price change direction consistency ranking. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] values = data.value # close prices last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._delta_window + self._consistency_window if len(values.shape) > 1 and len(values) >= min_len: # close_delta = ts_delta(close, delta_window) if len(values) > self._delta_window: close_delta = values[self._delta_window:] - values[:-self._delta_window] else: close_delta = np.zeros_like(values) if len(close_delta) >= self._consistency_window: # min_delta = ts_min(close_delta, consistency_window) window_deltas = close_delta[-self._consistency_window:] min_delta = np.min(window_deltas, axis=0) # max_delta = ts_max(close_delta, consistency_window) max_delta = np.max(window_deltas, axis=0) # Current close_delta current_delta = close_delta[-1] # cond1 = min_delta > 0 cond1 = min_delta > 0 # cond2 = max_delta < 0 cond2 = max_delta < 0 # neg_close_delta = close_delta * -1 neg_close_delta = -current_delta # inner_condition = condition(cond2, close_delta, neg_close_delta) inner_condition = np.where(cond2, current_delta, neg_close_delta) # condition_result = condition(cond1, close_delta, inner_condition) condition_result = np.where(cond1, current_delta, inner_condition) # rank(condition_result) cross-sectionally compute_mask = exists & valid ranked = np.zeros(n_symbols) if compute_mask.any(): valid_vals = condition_result[compute_mask] ranks = (np.argsort(np.argsort(valid_vals)) + 1) / len(valid_vals) ranked[compute_mask] = ranks result[compute_mask] = ranked[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #011: VWAP-close deviation and volume change signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._window: # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # vwap_close_diff = vwap - close vwap_close_diff = vwap - close # max_diff and min_diff over window window_diff = vwap_close_diff[-self._window:] max_diff = np.max(window_diff, axis=0) min_diff = np.min(window_diff, axis=0) # volume_delta = ts_delta(volume, window) if len(volume) > self._window: volume_delta = volume[-1] - volume[-(self._window + 1)] else: volume_delta = np.zeros(n_symbols) compute_mask = exists & valid if compute_mask.any(): # rank(max_diff) ranked_max = np.zeros(n_symbols) valid_max = max_diff[compute_mask] ranks_max = (np.argsort(np.argsort(valid_max)) + 1) / len(valid_max) ranked_max[compute_mask] = ranks_max # rank(min_diff) ranked_min = np.zeros(n_symbols) valid_min = min_diff[compute_mask] ranks_min = (np.argsort(np.argsort(valid_min)) + 1) / len(valid_min) ranked_min[compute_mask] = ranks_min # first_part = rank(max_diff) + rank(min_diff) first_part = ranked_max + ranked_min # rank(volume_delta) ranked_vol = np.zeros(n_symbols) valid_vol = volume_delta[compute_mask] ranks_vol = (np.argsort(np.argsort(valid_vol)) + 1) / len(valid_vol) ranked_vol[compute_mask] = ranks_vol # alpha = first_part * rank(volume_delta) alpha = first_part * ranked_vol result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #012: Volume-price divergence signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data[0], data[1] else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) > self._delta_window: # volume_delta = ts_delta(volume, delta_window) volume_delta = volume[-1] - volume[-(self._delta_window + 1)] # volume_sign = sign(volume_delta) volume_sign = np.sign(volume_delta) # close_delta = ts_delta(close, delta_window) close_delta = close[-1] - close[-(self._delta_window + 1)] # alpha = sign(volume_delta) * (-close_delta) alpha = volume_sign * (-close_delta) compute_mask = exists & valid result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #013: Close-volume covariance signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data[0], data[1] else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._cov_window: compute_mask = exists & valid cov_values = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: close_series = close[-self._cov_window:, i] vol_series = volume[-self._cov_window:, i] valid_mask = ~(np.isnan(close_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: close_valid = close_series[valid_mask] vol_valid = vol_series[valid_mask] # Covariance cov = np.cov(close_valid, vol_valid)[0, 1] cov_values[i] = cov # rank(cov) cross-sectionally if compute_mask.any(): ranked_cov = np.zeros(n_symbols) valid_cov = cov_values[compute_mask] # Handle NaN in ranking valid_cov = np.nan_to_num(valid_cov, nan=0.0) ranks = (np.argsort(np.argsort(valid_cov)) + 1) / len(valid_cov) ranked_cov[compute_mask] = ranks # alpha = -rank(cov) alpha = -ranked_cov result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #014: Returns delta and open-volume correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, close_data, volume_data = data else: open_data = close_data = volume_data = data open_ = open_data.value close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._delta_window, self._corr_window) + 1 if len(close.shape) > 1 and len(close) >= min_len: # returns = ts_returns(close) with np.errstate(divide='ignore', invalid='ignore'): returns = np.diff(close, axis=0) / close[:-1] # returns_delta = ts_delta(returns, delta_window) if len(returns) > self._delta_window: returns_delta = returns[-1] - returns[-(self._delta_window + 1)] else: returns_delta = np.zeros(n_symbols) compute_mask = exists & valid # rank(returns_delta) and negate ranked_returns = np.zeros(n_symbols) if compute_mask.any(): valid_returns = returns_delta[compute_mask] valid_returns = np.nan_to_num(valid_returns, nan=0.0) ranks = (np.argsort(np.argsort(valid_returns)) + 1) / len(valid_returns) ranked_returns[compute_mask] = ranks neg_ranked_returns = -ranked_returns # open_volume_corr = ts_corr(open, volume, corr_window) open_vol_corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: open_series = open_[-self._corr_window:, i] vol_series = volume[-self._corr_window:, i] valid_mask = ~(np.isnan(open_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: open_valid = open_series[valid_mask] vol_valid = vol_series[valid_mask] if np.std(open_valid) > 0 and np.std(vol_valid) > 0: corr = np.corrcoef(open_valid, vol_valid)[0, 1] open_vol_corr[i] = corr if not np.isnan(corr) else 0.0 # alpha = neg_rank_returns * open_volume_corr alpha = neg_ranked_returns * open_vol_corr result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #015: High-volume correlation ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, volume_data = data[0], data[1] else: high_data = volume_data = data high = high_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._corr_window + self._sum_window if len(high.shape) > 1 and len(high) >= min_len: compute_mask = exists & valid # Calculate rolling correlations for sum_window periods corr_ranks = [] for t in range(self._sum_window): offset = self._sum_window - 1 - t end_idx = len(high) - offset if offset > 0 else len(high) start_idx = end_idx - self._corr_window if start_idx >= 0: corr_values = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: high_series = high[start_idx:end_idx, i] vol_series = volume[start_idx:end_idx, i] valid_mask = ~(np.isnan(high_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: high_valid = high_series[valid_mask] vol_valid = vol_series[valid_mask] if np.std(high_valid) > 0 and np.std(vol_valid) > 0: corr = np.corrcoef(high_valid, vol_valid)[0, 1] corr_values[i] = corr if not np.isnan(corr) else 0.0 # rank(corr) cross-sectionally ranked_corr = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr_values[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) ranked_corr[compute_mask] = ranks corr_ranks.append(ranked_corr) if corr_ranks: # sum_rank = ts_sum(corr_rank, sum_window) sum_rank = np.sum(corr_ranks, axis=0) # alpha = -sum_rank alpha = -sum_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #016: High-volume covariance ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, volume_data = data[0], data[1] else: high_data = volume_data = data high = high_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(high.shape) > 1 and len(high) >= self._cov_window: compute_mask = exists & valid cov_values = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: high_series = high[-self._cov_window:, i] vol_series = volume[-self._cov_window:, i] valid_mask = ~(np.isnan(high_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: high_valid = high_series[valid_mask] vol_valid = vol_series[valid_mask] cov = np.cov(high_valid, vol_valid)[0, 1] cov_values[i] = cov if not np.isnan(cov) else 0.0 # rank(cov) and negate if compute_mask.any(): ranked_cov = np.zeros(n_symbols) valid_cov = cov_values[compute_mask] valid_cov = np.nan_to_num(valid_cov, nan=0.0) ranks = (np.argsort(np.argsort(valid_cov)) + 1) / len(valid_cov) ranked_cov[compute_mask] = ranks alpha = -ranked_cov result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #017: Complex close-volume momentum signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data[0], data[1] else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._ts_rank_window1, self._ts_rank_window2, self._amount_mean_window) + 2 if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # ts_rank(close, ts_rank_window1) for each symbol close_ts_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: series = close[-self._ts_rank_window1:, i] valid_mask = ~np.isnan(series) if valid_mask.sum() >= 2: valid_series = series[valid_mask] current_val = valid_series[-1] close_ts_rank[i] = np.sum(valid_series <= current_val) / len(valid_series) # rank(close_ts_rank) cross-sectionally close_rank = np.zeros(n_symbols) if compute_mask.any(): valid_vals = close_ts_rank[compute_mask] ranks = (np.argsort(np.argsort(valid_vals)) + 1) / len(valid_vals) close_rank[compute_mask] = ranks neg_close_rank = -close_rank # close_delta2 = ts_delta(ts_delta(close, 1), 1) if len(close) >= 3: close_delta1 = close[1:] - close[:-1] close_delta2 = close_delta1[1:] - close_delta1[:-1] current_delta2 = close_delta2[-1] else: current_delta2 = np.zeros(n_symbols) # rank(close_delta2) delta_rank = np.zeros(n_symbols) if compute_mask.any(): valid_delta = current_delta2[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) delta_rank[compute_mask] = ranks # amount = volume * close amount = volume * close amount_mean = np.mean(amount[-self._amount_mean_window:], axis=0) # volume_ratio = volume / amount_mean with np.errstate(divide='ignore', invalid='ignore'): volume_ratio = volume[-1] / amount_mean # ts_rank(volume_ratio, ts_rank_window2) volume_ts_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: # Calculate volume_ratio over time for ts_rank vol_ratios = volume[-self._ts_rank_window2:, i] / amount_mean[i] valid_mask = ~(np.isnan(vol_ratios) | np.isinf(vol_ratios)) if valid_mask.sum() >= 2: valid_series = vol_ratios[valid_mask] current_val = valid_series[-1] volume_ts_rank[i] = np.sum(valid_series <= current_val) / len(valid_series) # rank(volume_ts_rank) volume_rank = np.zeros(n_symbols) if compute_mask.any(): valid_vol = volume_ts_rank[compute_mask] valid_vol = np.nan_to_num(valid_vol, nan=0.0) ranks = (np.argsort(np.argsort(valid_vol)) + 1) / len(valid_vol) volume_rank[compute_mask] = ranks # alpha = neg_close_rank * delta_rank * volume_rank alpha = neg_close_rank * delta_rank * volume_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #018: Close-open volatility and correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, open_data = data[0], data[1] else: close_data = open_data = data close = close_data.value open_ = open_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._std_window, self._corr_window) if len(close.shape) > 1 and len(close) >= min_len: # close_open_diff = close - open close_open_diff = close - open_ # abs_diff = abs(close_open_diff) abs_diff = np.abs(close_open_diff) # std = ts_std(abs_diff, std_window) std_values = np.std(abs_diff[-self._std_window:], axis=0) # corr = ts_corr(close, open, corr_window) compute_mask = exists & valid corr_values = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: close_series = close[-self._corr_window:, i] open_series = open_[-self._corr_window:, i] valid_mask = ~(np.isnan(close_series) | np.isnan(open_series)) if valid_mask.sum() >= 3: close_valid = close_series[valid_mask] open_valid = open_series[valid_mask] if np.std(close_valid) > 0 and np.std(open_valid) > 0: corr = np.corrcoef(close_valid, open_valid)[0, 1] corr_values[i] = corr if not np.isnan(corr) else 0.0 # Current close_open_diff current_diff = close_open_diff[-1] # sum_all = std + close_open_diff + corr sum_all = std_values + current_diff + corr_values # rank(sum_all) and negate if compute_mask.any(): ranked = np.zeros(n_symbols) valid_sum = sum_all[compute_mask] valid_sum = np.nan_to_num(valid_sum, nan=0.0) ranks = (np.argsort(np.argsort(valid_sum)) + 1) / len(valid_sum) ranked[compute_mask] = ranks alpha = -ranked result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #019: Price direction and long-term returns signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] values = data.value # close prices last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._delay_window, self._delta_window, self._returns_sum_window) + 1 if len(values.shape) > 1 and len(values) >= min_len: close = values # close_lag = delay(close, delay_window) close_lag = close[-(self._delay_window + 1)] # close_diff = close - close_lag close_diff = close[-1] - close_lag # close_delta = ts_delta(close, delta_window) close_delta = close[-1] - close[-(self._delta_window + 1)] # sum_changes = close_diff + close_delta sum_changes = close_diff + close_delta # sign_changes = sign(sum_changes) sign_changes = np.sign(sum_changes) # neg_sign = -sign_changes neg_sign = -sign_changes # returns = ts_returns(close) with np.errstate(divide='ignore', invalid='ignore'): returns = np.diff(close, axis=0) / close[:-1] # returns_sum = ts_sum(returns, returns_sum_window) window = min(self._returns_sum_window, len(returns)) returns_sum = np.sum(returns[-window:], axis=0) # returns_plus1 = returns_sum + 1 returns_plus1 = returns_sum + 1 # rank(returns_plus1) compute_mask = exists & valid returns_rank = np.zeros(n_symbols) if compute_mask.any(): valid_returns = returns_plus1[compute_mask] valid_returns = np.nan_to_num(valid_returns, nan=0.0) ranks = (np.argsort(np.argsort(valid_returns)) + 1) / len(valid_returns) returns_rank[compute_mask] = ranks # rank_plus1 = returns_rank + 1 rank_plus1 = returns_rank + 1 # alpha = neg_sign * rank_plus1 alpha = neg_sign * rank_plus1 result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #020: Opening gap ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data = data else: open_data = high_data = low_data = close_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(open_.shape) > 1 and len(open_) > self._delay_window: # Lagged values high_lag = high[-(self._delay_window + 1)] close_lag = close[-(self._delay_window + 1)] low_lag = low[-(self._delay_window + 1)] # Current open current_open = open_[-1] # open_high_diff = open - high_lag open_high_diff = current_open - high_lag # open_close_diff = open - close_lag open_close_diff = current_open - close_lag # open_low_diff = open - low_lag open_low_diff = current_open - low_lag compute_mask = exists & valid if compute_mask.any(): # rank(open_high_diff) ranked_oh = np.zeros(n_symbols) valid_oh = open_high_diff[compute_mask] valid_oh = np.nan_to_num(valid_oh, nan=0.0) ranks = (np.argsort(np.argsort(valid_oh)) + 1) / len(valid_oh) ranked_oh[compute_mask] = ranks # neg_open_high_rank = -rank(open_high_diff) neg_ranked_oh = -ranked_oh # rank(open_close_diff) ranked_oc = np.zeros(n_symbols) valid_oc = open_close_diff[compute_mask] valid_oc = np.nan_to_num(valid_oc, nan=0.0) ranks = (np.argsort(np.argsort(valid_oc)) + 1) / len(valid_oc) ranked_oc[compute_mask] = ranks # rank(open_low_diff) ranked_ol = np.zeros(n_symbols) valid_ol = open_low_diff[compute_mask] valid_ol = np.nan_to_num(valid_ol, nan=0.0) ranks = (np.argsort(np.argsort(valid_ol)) + 1) / len(valid_ol) ranked_ol[compute_mask] = ranks # alpha = neg_open_high_rank * open_close_rank * open_low_rank alpha = neg_ranked_oh * ranked_oc * ranked_ol result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #021: Close volatility and volume ratio signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data[0], data[1] else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._long_window, self._amount_window) if len(close.shape) > 1 and len(close) >= min_len: # amount = volume * close amount = volume * close # close_mean_8 = ts_sum(close, 8) / 8 close_mean_8 = np.mean(close[-self._long_window:], axis=0) # close_std_8 = ts_std(close, 8) close_std_8 = np.std(close[-self._long_window:], axis=0) # close_mean_2 = ts_sum(close, 2) / 2 close_mean_2 = np.mean(close[-self._short_window:], axis=0) # amount_mean = ts_mean(amount, 20) amount_mean = np.mean(amount[-self._amount_window:], axis=0) # volume_ratio = volume / amount_mean with np.errstate(divide='ignore', invalid='ignore'): volume_ratio = volume[-1] / amount_mean # condition1: close_mean_8 + close_std_8 < close_mean_2 condition1 = (close_mean_8 + close_std_8) < close_mean_2 # condition2: close_mean_2 < close_mean_8 - close_std_8 condition2 = close_mean_2 < (close_mean_8 - close_std_8) # condition3: volume_ratio >= 1 condition3 = volume_ratio >= 1 # inner_condition = condition(condition3, 1, -1) inner_condition = np.where(condition3, 1.0, -1.0) # middle_condition = condition(condition2, 1, inner_condition) middle_condition = np.where(condition2, 1.0, inner_condition) # alpha = condition(condition1, -1, middle_condition) alpha = np.where(condition1, -1.0, middle_condition) compute_mask = exists & valid result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #022: High-volume correlation change signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, close_data, volume_data = data else: high_data = close_data = volume_data = data high = high_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._corr_window + self._delta_window, self._std_window) if len(high.shape) > 1 and len(high) >= min_len: compute_mask = exists & valid # Calculate correlation at current and lagged time corr_current = np.zeros(n_symbols) corr_lagged = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: # Current correlation high_series = high[-self._corr_window:, i] vol_series = volume[-self._corr_window:, i] valid_mask = ~(np.isnan(high_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: if np.std(high_series[valid_mask]) > 0 and np.std(vol_series[valid_mask]) > 0: corr_current[i] = np.corrcoef(high_series[valid_mask], vol_series[valid_mask])[0, 1] # Lagged correlation end_idx = -(self._delta_window) start_idx = end_idx - self._corr_window if start_idx >= -len(high): high_lag = high[start_idx:end_idx, i] vol_lag = volume[start_idx:end_idx, i] valid_mask = ~(np.isnan(high_lag) | np.isnan(vol_lag)) if valid_mask.sum() >= 3: if np.std(high_lag[valid_mask]) > 0 and np.std(vol_lag[valid_mask]) > 0: corr_lagged[i] = np.corrcoef(high_lag[valid_mask], vol_lag[valid_mask])[0, 1] # corr_delta = corr_current - corr_lagged corr_delta = corr_current - corr_lagged # close_std = ts_std(close, std_window) close_std = np.std(close[-self._std_window:], axis=0) # rank(close_std) std_rank = np.zeros(n_symbols) if compute_mask.any(): valid_std = close_std[compute_mask] valid_std = np.nan_to_num(valid_std, nan=0.0) ranks = (np.argsort(np.argsort(valid_std)) + 1) / len(valid_std) std_rank[compute_mask] = ranks # alpha = -(corr_delta * std_rank) alpha = -(corr_delta * std_rank) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #023: High price breakout signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] values = data.value # high prices last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._avg_window, self._delta_window + 1) if len(values.shape) > 1 and len(values) >= min_len: high = values # high_mean = ts_sum(high, avg_window) / avg_window high_mean = np.mean(high[-self._avg_window:], axis=0) # condition = high_mean < high (current) condition = high_mean < high[-1] # high_delta = ts_delta(high, delta_window) high_delta = high[-1] - high[-(self._delta_window + 1)] # neg_delta = -high_delta neg_delta = -high_delta # alpha = condition(condition, neg_delta, 0) alpha = np.where(condition, neg_delta, 0.0) compute_mask = exists & valid result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #024: Long-term average change rate signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] values = data.value # close prices last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(values.shape) > 1 and len(values) >= self._long_window + 1: close = values # close_mean = ts_sum(close, window) / window close_mean_current = np.mean(close[-self._long_window:], axis=0) # mean_delta = ts_delta(close_mean, window) # This requires calculating close_mean at lagged position close_mean_lagged = np.mean(close[-(2 * self._long_window):-self._long_window], axis=0) mean_delta = close_mean_current - close_mean_lagged # close_lag = delay(close, window) close_lag = close[-self._long_window - 1] # rate = mean_delta / close_lag with np.errstate(divide='ignore', invalid='ignore'): rate = mean_delta / close_lag # main_condition = rate <= threshold main_condition = rate <= self._threshold # close_min = ts_min(close, window) close_min = np.min(close[-self._long_window:], axis=0) # close_min_diff = close - close_min close_min_diff = close[-1] - close_min # neg_close_min = -close_min_diff neg_close_min = -close_min_diff # close_delta = ts_delta(close, short_delta) close_delta = close[-1] - close[-(self._short_delta + 1)] # neg_delta = -close_delta neg_delta = -close_delta # alpha = condition(main_condition, neg_close_min, neg_delta) alpha = np.where(main_condition, neg_close_min, neg_delta) compute_mask = exists & valid result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #025: Returns-amount-VWAP composite signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._amount_window + 1: # returns = ts_returns(close) with np.errstate(divide='ignore', invalid='ignore'): returns = np.diff(close, axis=0) / close[:-1] current_returns = returns[-1] # neg_returns = -returns neg_returns = -current_returns # amount = volume * close amount = volume * close # amount_mean = ts_mean(amount, amount_window) amount_mean = np.mean(amount[-self._amount_window:], axis=0) # vwap = (high + low + close) / 3 * volume / sum(volume) (cumulative) typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) current_vwap = vwap[-1] # high_close_diff = high - close high_close_diff = high[-1] - close[-1] # product = neg_returns * amount_mean * vwap * high_close_diff product = neg_returns * amount_mean * current_vwap * high_close_diff # rank(product) compute_mask = exists & valid ranked = np.zeros(n_symbols) if compute_mask.any(): valid_product = product[compute_mask] valid_product = np.nan_to_num(valid_product, nan=0.0) ranks = (np.argsort(np.argsort(valid_product)) + 1) / len(valid_product) ranked[compute_mask] = ranks result[compute_mask] = ranked[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #026: Volume-high time-series correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): volume_data, high_data = data[0], data[1] else: volume_data = high_data = data volume = volume_data.value high = high_data.value last = volume_data[-1] if len(volume_data) > 0 else volume_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._ts_rank_window + self._corr_window + self._max_window if len(volume.shape) > 1 and len(volume) >= min_len: compute_mask = exists & valid # Calculate correlations for max_window periods corr_values = [] for t in range(self._max_window): offset = self._max_window - 1 - t end_idx = len(volume) - offset if offset > 0 else len(volume) corr_at_t = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: # ts_rank for volume and high vol_window = volume[end_idx - self._ts_rank_window - self._corr_window:end_idx, i] high_window = high[end_idx - self._ts_rank_window - self._corr_window:end_idx, i] # Calculate ts_rank series vol_ranks = [] high_ranks = [] for j in range(self._corr_window): idx = self._ts_rank_window + j vol_slice = vol_window[:idx + 1] high_slice = high_window[:idx + 1] if len(vol_slice) >= 2: vol_rank = np.sum(vol_slice <= vol_slice[-1]) / len(vol_slice) high_rank = np.sum(high_slice <= high_slice[-1]) / len(high_slice) vol_ranks.append(vol_rank) high_ranks.append(high_rank) if len(vol_ranks) >= 3: vol_ranks = np.array(vol_ranks) high_ranks = np.array(high_ranks) if np.std(vol_ranks) > 0 and np.std(high_ranks) > 0: corr_at_t[i] = np.corrcoef(vol_ranks, high_ranks)[0, 1] corr_values.append(corr_at_t) if corr_values: # max_corr = ts_max(corr, max_window) corr_array = np.array(corr_values) max_corr = np.max(corr_array, axis=0) # alpha = -max_corr alpha = -max_corr result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #027: Volume-VWAP correlation ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): volume_data, high_data, low_data, close_data = data else: volume_data = high_data = low_data = close_data = data volume = volume_data.value high = high_data.value low = low_data.value close = close_data.value last = volume_data[-1] if len(volume_data) > 0 else volume_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._corr_window + self._sum_window if len(volume.shape) > 1 and len(volume) >= min_len: compute_mask = exists & valid # Calculate VWAP = (high + low + close) / 3 * volume / volume # Simplified: typical price = (high + low + close) / 3 typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Cross-sectional rank of volume and vwap at each time def cross_rank(arr): ranked = np.zeros_like(arr) for t in range(len(arr)): row = arr[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: ranks = (np.argsort(np.argsort(np.where(valid_mask, row, 0))) + 1) / valid_mask.sum() ranked[t] = np.where(valid_mask, ranks, np.nan) return ranked volume_rank = cross_rank(volume) vwap_rank = cross_rank(vwap) # Calculate correlation over corr_window for sum_window periods corr_sum = np.zeros(n_symbols) for t in range(self._sum_window): offset = self._sum_window - 1 - t end_idx = len(volume) - offset if offset > 0 else len(volume) start_idx = end_idx - self._corr_window if start_idx >= 0: for i in range(n_symbols): if compute_mask[i]: vol_series = volume_rank[start_idx:end_idx, i] vwap_series = vwap_rank[start_idx:end_idx, i] valid_mask = ~(np.isnan(vol_series) | np.isnan(vwap_series)) if valid_mask.sum() >= 3: if np.std(vol_series[valid_mask]) > 0 and np.std(vwap_series[valid_mask]) > 0: corr = np.corrcoef(vol_series[valid_mask], vwap_series[valid_mask])[0, 1] corr_sum[i] += corr # div_result = sum_corr / 2.0 div_result = corr_sum / 2.0 # rank(div_result) rank_result = np.zeros(n_symbols) if compute_mask.any(): valid_div = div_result[compute_mask] valid_div = np.nan_to_num(valid_div, nan=0.0) ranks = (np.argsort(np.argsort(valid_div)) + 1) / len(valid_div) rank_result[compute_mask] = ranks # condition: 0.5 < rank_result -> -1, else 1 alpha = np.where(rank_result > 0.5, -1.0, 1.0) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #028: Amount-low correlation with mid-price signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._amount_window + self._corr_window if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # amount = volume * close amount = volume * close # amount_mean = ts_mean(amount, amount_window) amount_mean = np.zeros_like(amount) for t in range(self._amount_window - 1, len(amount)): amount_mean[t] = np.mean(amount[t - self._amount_window + 1:t + 1], axis=0) # corr = ts_corr(amount_mean, low, corr_window) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: am_series = amount_mean[-self._corr_window:, i] low_series = low[-self._corr_window:, i] valid_mask = ~(np.isnan(am_series) | np.isnan(low_series)) if valid_mask.sum() >= 3: if np.std(am_series[valid_mask]) > 0 and np.std(low_series[valid_mask]) > 0: corr[i] = np.corrcoef(am_series[valid_mask], low_series[valid_mask])[0, 1] # mid_price = (high + low) / 2 mid_price = (high[-1] + low[-1]) / 2 # sum_part = corr + mid_price sum_part = corr + mid_price # diff = sum_part - close diff = sum_part - close[-1] # scale cross-sectionally if compute_mask.any(): valid_diff = diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) diff_std = np.std(valid_diff) if diff_std > 0: alpha = (diff - np.mean(valid_diff)) / diff_std else: alpha = np.zeros(n_symbols) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #029: Complex nested ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] close = data.value last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._delta_window + self._delay_window + self._ts_rank_window + 2 if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # returns = (close - close_lag1) / close_lag1 returns = np.zeros_like(close) returns[1:] = (close[1:] - close[:-1]) / np.where(close[:-1] != 0, close[:-1], 1) # close_minus1 = close - 1 close_minus1 = close - 1 # delta = ts_delta(close_minus1, delta_window) delta = close_minus1[-1] - close_minus1[-(self._delta_window + 1)] # rank1 = rank(delta) rank1 = np.zeros(n_symbols) if compute_mask.any(): valid_delta = delta[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) rank1[compute_mask] = ranks # neg_rank = -rank1 neg_rank = -rank1 # rank2 = rank(neg_rank) rank2 = np.zeros(n_symbols) if compute_mask.any(): valid_neg = neg_rank[compute_mask] valid_neg = np.nan_to_num(valid_neg, nan=0.0) ranks = (np.argsort(np.argsort(valid_neg)) + 1) / len(valid_neg) rank2[compute_mask] = ranks # rank3 = rank(rank2) rank3 = np.zeros(n_symbols) if compute_mask.any(): valid_r2 = rank2[compute_mask] valid_r2 = np.nan_to_num(valid_r2, nan=0.0) ranks = (np.argsort(np.argsort(valid_r2)) + 1) / len(valid_r2) rank3[compute_mask] = ranks # min_2 = ts_min(rank3, 2) - simplified as rank3 since we only have current min_2 = rank3 # sum_1 = ts_sum(min_2, 1) = min_2 sum_1 = min_2 # log_result = log(sum_1) with np.errstate(divide='ignore', invalid='ignore'): log_result = np.log(np.maximum(sum_1, 1e-10)) # scaled = scale(log_result) scaled = np.zeros(n_symbols) if compute_mask.any(): valid_log = log_result[compute_mask] valid_log = np.nan_to_num(valid_log, nan=0.0) log_std = np.std(valid_log) if log_std > 0: scaled[compute_mask] = (valid_log - np.mean(valid_log)) / log_std # rank4 = rank(scaled) rank4 = np.zeros(n_symbols) if compute_mask.any(): valid_scaled = scaled[compute_mask] valid_scaled = np.nan_to_num(valid_scaled, nan=0.0) ranks = (np.argsort(np.argsort(valid_scaled)) + 1) / len(valid_scaled) rank4[compute_mask] = ranks # rank5 = rank(rank4) rank5 = np.zeros(n_symbols) if compute_mask.any(): valid_r4 = rank4[compute_mask] valid_r4 = np.nan_to_num(valid_r4, nan=0.0) ranks = (np.argsort(np.argsort(valid_r4)) + 1) / len(valid_r4) rank5[compute_mask] = ranks # product = ts_product(rank5, ts_rank_window) - simplified product = rank5 ** self._ts_rank_window # neg_returns = -returns neg_returns = -returns # delayed_returns = delay(neg_returns, delay_window) delayed_returns = neg_returns[-(self._delay_window + 1)] # rank_returns = ts_rank(delayed_returns, ts_rank_window) rank_returns = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: series = neg_returns[-(self._delay_window + self._ts_rank_window):-(self._delay_window), i] if len(series) >= 2: current_val = delayed_returns[i] rank_returns[i] = np.sum(series <= current_val) / len(series) # alpha = min(product, rank_returns) alpha = np.minimum(product, rank_returns) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #030: Price direction pattern with volume ratio signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._long_volume_window, 4) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # close_lag1 = delay(close, 1) close_lag1 = close[-2] # close_lag2 = delay(close, 2) close_lag2 = close[-3] # close_lag3 = delay(close, 3) close_lag3 = close[-4] # sign1 = sign(close - close_lag1) sign1 = np.sign(close[-1] - close_lag1) # sign2 = sign(close_lag1 - close_lag2) sign2 = np.sign(close_lag1 - close_lag2) # sign3 = sign(close_lag2 - close_lag3) sign3 = np.sign(close_lag2 - close_lag3) # sign_sum = sign1 + sign2 + sign3 sign_sum = sign1 + sign2 + sign3 # sign_rank = rank(sign_sum) sign_rank = np.zeros(n_symbols) if compute_mask.any(): valid_sign = sign_sum[compute_mask] valid_sign = np.nan_to_num(valid_sign, nan=0.0) ranks = (np.argsort(np.argsort(valid_sign)) + 1) / len(valid_sign) sign_rank[compute_mask] = ranks # one_minus_rank = 1.0 - sign_rank one_minus_rank = 1.0 - sign_rank # volume_sum_5 = ts_sum(volume, short_volume_window) volume_sum_5 = np.sum(volume[-self._short_volume_window:], axis=0) # volume_sum_20 = ts_sum(volume, long_volume_window) volume_sum_20 = np.sum(volume[-self._long_volume_window:], axis=0) # numerator = one_minus_rank * volume_sum_5 numerator = one_minus_rank * volume_sum_5 # alpha = numerator / volume_sum_20 with np.errstate(divide='ignore', invalid='ignore'): alpha = numerator / np.where(volume_sum_20 != 0, volume_sum_20, 1) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #031: Multi-ranking decay + amount-low correlation. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, low_data, volume_data = data else: close_data = low_data = volume_data = data close = close_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._long_delta + self._decay_window, self._amount_window + self._corr_window) + 1 if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # amount = volume * close amount = volume * close # Part 1: rank(rank(ts_delta(close, 10))) close_delta10 = close[-1] - close[-(self._long_delta + 1)] rank1 = np.zeros(n_symbols) if compute_mask.any(): valid_delta = close_delta10[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) rank1[compute_mask] = ranks # rank2 = rank(rank1) rank2 = np.zeros(n_symbols) if compute_mask.any(): valid_r1 = rank1[compute_mask] ranks = (np.argsort(np.argsort(valid_r1)) + 1) / len(valid_r1) rank2[compute_mask] = ranks # neg_rank = -rank2 neg_rank = -rank2 # ts_decayed_linear: weighted sum with decaying weights weights = np.arange(self._decay_window, 0, -1, dtype=float) weights = weights / weights.sum() decayed = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: decayed[i] = neg_rank[i] # rank3 = rank(decayed) rank3 = np.zeros(n_symbols) if compute_mask.any(): valid_dec = decayed[compute_mask] valid_dec = np.nan_to_num(valid_dec, nan=0.0) ranks = (np.argsort(np.argsort(valid_dec)) + 1) / len(valid_dec) rank3[compute_mask] = ranks # rank4 = rank(rank3) rank4 = np.zeros(n_symbols) if compute_mask.any(): valid_r3 = rank3[compute_mask] ranks = (np.argsort(np.argsort(valid_r3)) + 1) / len(valid_r3) rank4[compute_mask] = ranks # first_part = rank(rank4) first_part = np.zeros(n_symbols) if compute_mask.any(): valid_r4 = rank4[compute_mask] ranks = (np.argsort(np.argsort(valid_r4)) + 1) / len(valid_r4) first_part[compute_mask] = ranks # Part 2: rank(-ts_delta(close, 3)) close_delta3 = close[-1] - close[-(self._short_delta + 1)] neg_delta3 = -close_delta3 second_part = np.zeros(n_symbols) if compute_mask.any(): valid_neg = neg_delta3[compute_mask] valid_neg = np.nan_to_num(valid_neg, nan=0.0) ranks = (np.argsort(np.argsort(valid_neg)) + 1) / len(valid_neg) second_part[compute_mask] = ranks # Part 3: sign(scale(ts_corr(amount_mean, low, corr_window))) amount_mean = np.mean(amount[-self._amount_window:], axis=0) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: am_series = np.zeros(self._corr_window) for t in range(self._corr_window): idx = len(amount) - self._corr_window + t am_series[t] = np.mean(amount[max(0, idx - self._amount_window + 1):idx + 1, i]) low_series = low[-self._corr_window:, i] valid_mask = ~(np.isnan(am_series) | np.isnan(low_series)) if valid_mask.sum() >= 3: if np.std(am_series[valid_mask]) > 0 and np.std(low_series[valid_mask]) > 0: corr[i] = np.corrcoef(am_series[valid_mask], low_series[valid_mask])[0, 1] scaled_corr = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) corr_std = np.std(valid_corr) if corr_std > 0: scaled_corr[compute_mask] = (valid_corr - np.mean(valid_corr)) / corr_std third_part = np.sign(scaled_corr) # alpha = first_part + second_part + third_part alpha = first_part + second_part + third_part result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #032: Mean reversion with VWAP correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._mean_window, self._corr_window + self._delay_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Part 1: scale(close_mean - close) close_mean = np.mean(close[-self._mean_window:], axis=0) diff = close_mean - close[-1] first_part = np.zeros(n_symbols) if compute_mask.any(): valid_diff = diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) diff_std = np.std(valid_diff) if diff_std > 0: first_part[compute_mask] = (valid_diff - np.mean(valid_diff)) / diff_std # Part 2: 20 * scale(ts_corr(vwap, delay(close, 5), 230)) typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vwap_end = len(vwap) close_end = len(close) - self._delay_window if close_end >= self._corr_window: vwap_series = vwap[vwap_end - self._corr_window:vwap_end, i] close_series = close[close_end - self._corr_window:close_end, i] valid_mask = ~(np.isnan(vwap_series) | np.isnan(close_series)) if valid_mask.sum() >= 3: if np.std(vwap_series[valid_mask]) > 0 and np.std(close_series[valid_mask]) > 0: corr[i] = np.corrcoef(vwap_series[valid_mask], close_series[valid_mask])[0, 1] scaled_corr = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) corr_std = np.std(valid_corr) if corr_std > 0: scaled_corr[compute_mask] = (valid_corr - np.mean(valid_corr)) / corr_std second_part = 20 * scaled_corr # alpha = first_part + second_part alpha = first_part + second_part result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #033: Open-close ratio momentum signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, close_data = data else: open_data = close_data = data open_ = open_data.value close = close_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(open_.shape) > 1 and len(open_) >= 1: compute_mask = exists & valid # open / close with np.errstate(divide='ignore', invalid='ignore'): open_close_ratio = open_[-1] / np.where(close[-1] != 0, close[-1], 1) # 1 - (open / close) one_minus_ratio = 1 - open_close_ratio # -1 * one_minus_ratio neg_powered = -one_minus_ratio # rank(neg_powered) alpha = np.zeros(n_symbols) if compute_mask.any(): valid_val = neg_powered[compute_mask] valid_val = np.nan_to_num(valid_val, nan=0.0) ranks = (np.argsort(np.argsort(valid_val)) + 1) / len(valid_val) alpha[compute_mask] = ranks result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #034: Volatility ratio and price change signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] close = data.value last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._long_std, self._delta_window) + 2 if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate returns returns = np.zeros_like(close) returns[1:] = (close[1:] - close[:-1]) / np.where(close[:-1] != 0, close[:-1], 1) # Part 1: 1 - rank(ts_std(returns, short) / ts_std(returns, long)) returns_std_short = np.std(returns[-self._short_std:], axis=0) returns_std_long = np.std(returns[-self._long_std:], axis=0) with np.errstate(divide='ignore', invalid='ignore'): std_ratio = returns_std_short / np.where(returns_std_long != 0, returns_std_long, 1) std_rank = np.zeros(n_symbols) if compute_mask.any(): valid_ratio = std_ratio[compute_mask] valid_ratio = np.nan_to_num(valid_ratio, nan=0.0) ranks = (np.argsort(np.argsort(valid_ratio)) + 1) / len(valid_ratio) std_rank[compute_mask] = ranks first_part = 1 - std_rank # Part 2: 1 - rank(ts_delta(close, 1)) close_delta = close[-1] - close[-(self._delta_window + 1)] delta_rank = np.zeros(n_symbols) if compute_mask.any(): valid_delta = close_delta[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) delta_rank[compute_mask] = ranks second_part = 1 - delta_rank # rank(first_part + second_part) sum_parts = first_part + second_part alpha = np.zeros(n_symbols) if compute_mask.any(): valid_sum = sum_parts[compute_mask] valid_sum = np.nan_to_num(valid_sum, nan=0.0) ranks = (np.argsort(np.argsort(valid_sum)) + 1) / len(valid_sum) alpha[compute_mask] = ranks result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #035: Volume-price-returns time-series ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._volume_window, self._price_window, self._returns_window) + 1 if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate returns returns = np.zeros_like(close) returns[1:] = (close[1:] - close[:-1]) / np.where(close[:-1] != 0, close[:-1], 1) # ts_rank(volume, volume_window) volume_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vol_series = volume[-self._volume_window:, i] current_vol = volume[-1, i] volume_rank[i] = np.sum(vol_series <= current_vol) / len(vol_series) # price_range = (close + high) - low price_range = (close + high) - low # ts_rank(price_range, price_window) price_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: pr_series = price_range[-self._price_window:, i] current_pr = price_range[-1, i] price_rank[i] = np.sum(pr_series <= current_pr) / len(pr_series) # 1 - price_rank one_minus_price = 1 - price_rank # ts_rank(returns, returns_window) returns_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: ret_series = returns[-self._returns_window:, i] current_ret = returns[-1, i] returns_rank[i] = np.sum(ret_series <= current_ret) / len(ret_series) # 1 - returns_rank one_minus_returns = 1 - returns_rank # alpha = volume_rank * one_minus_price * one_minus_returns alpha = volume_rank * one_minus_price * one_minus_returns result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #036: Weighted multi-factor composite signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, open_data, high_data, low_data, volume_data = data else: close_data = open_data = high_data = low_data = volume_data = data close = close_data.value open_ = open_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = 201 # Requires 200-day mean if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate returns and amount returns = np.zeros_like(close) returns[1:] = (close[1:] - close[:-1]) / np.where(close[:-1] != 0, close[:-1], 1) amount = volume * close # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Part 1: 2.21 * rank(ts_corr(close-open, delay(volume,1), 15)) close_open_diff = close - open_ volume_lag = volume[:-1] # delay by 1 corr1 = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: co_series = close_open_diff[-15:, i] vol_series = volume_lag[-15:, i] valid_mask = ~(np.isnan(co_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3 and np.std(co_series[valid_mask]) > 0 and np.std(vol_series[valid_mask]) > 0: corr1[i] = np.corrcoef(co_series[valid_mask], vol_series[valid_mask])[0, 1] rank1 = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr1[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) rank1[compute_mask] = ranks part1 = 2.21 * rank1 # Part 2: 0.7 * rank(open - close) open_close_diff = open_[-1] - close[-1] rank2 = np.zeros(n_symbols) if compute_mask.any(): valid_diff = open_close_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) rank2[compute_mask] = ranks part2 = 0.7 * rank2 # Part 3: 0.73 * rank(ts_rank(delay(-returns, 6), 5)) neg_returns = -returns ts_rank_values = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: # ts_rank of delayed neg_returns delayed_ret = neg_returns[:-6, i] if len(neg_returns) > 6 else neg_returns[:, i] if len(delayed_ret) >= 5: current_val = delayed_ret[-1] series = delayed_ret[-5:] ts_rank_values[i] = np.sum(series <= current_val) / len(series) rank3 = np.zeros(n_symbols) if compute_mask.any(): valid_ts = ts_rank_values[compute_mask] valid_ts = np.nan_to_num(valid_ts, nan=0.0) ranks = (np.argsort(np.argsort(valid_ts)) + 1) / len(valid_ts) rank3[compute_mask] = ranks part3 = 0.73 * rank3 # Part 4: rank(abs(ts_corr(vwap, ts_mean(amount, 20), 6))) amount_mean = np.zeros_like(amount) for t in range(19, len(amount)): amount_mean[t] = np.mean(amount[t - 19:t + 1], axis=0) corr4 = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vwap_series = vwap[-6:, i] am_series = amount_mean[-6:, i] valid_mask = ~(np.isnan(vwap_series) | np.isnan(am_series)) if valid_mask.sum() >= 3 and np.std(vwap_series[valid_mask]) > 0 and np.std(am_series[valid_mask]) > 0: corr4[i] = np.abs(np.corrcoef(vwap_series[valid_mask], am_series[valid_mask])[0, 1]) rank4 = np.zeros(n_symbols) if compute_mask.any(): valid_corr4 = corr4[compute_mask] valid_corr4 = np.nan_to_num(valid_corr4, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr4)) + 1) / len(valid_corr4) rank4[compute_mask] = ranks part4 = rank4 # Part 5: 0.6 * rank((close_mean_200 - open) * (close - open)) close_mean_200 = np.mean(close[-200:], axis=0) mean_open_diff = close_mean_200 - open_[-1] close_open_product = mean_open_diff * (close[-1] - open_[-1]) rank5 = np.zeros(n_symbols) if compute_mask.any(): valid_prod = close_open_product[compute_mask] valid_prod = np.nan_to_num(valid_prod, nan=0.0) ranks = (np.argsort(np.argsort(valid_prod)) + 1) / len(valid_prod) rank5[compute_mask] = ranks part5 = 0.6 * rank5 # alpha = part1 + part2 + part3 + part4 + part5 alpha = part1 + part2 + part3 + part4 + part5 result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #037: Long-term open-close correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, close_data = data else: open_data = close_data = data open_ = open_data.value close = close_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._corr_window + self._delay_window + 1 if len(open_.shape) > 1 and len(open_) >= min_len: compute_mask = exists & valid # open - close open_close_diff = open_ - close # delay(open_close_diff, 1) delayed_diff = open_close_diff[:-self._delay_window] if self._delay_window > 0 else open_close_diff # ts_corr(delayed_diff, close, corr_window) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: diff_series = delayed_diff[-self._corr_window:, i] close_series = close[-self._corr_window:, i] valid_mask = ~(np.isnan(diff_series) | np.isnan(close_series)) if valid_mask.sum() >= 3: if np.std(diff_series[valid_mask]) > 0 and np.std(close_series[valid_mask]) > 0: corr[i] = np.corrcoef(diff_series[valid_mask], close_series[valid_mask])[0, 1] # rank(corr) first_part = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) first_part[compute_mask] = ranks # rank(open_close_diff) current_diff = open_[-1] - close[-1] second_part = np.zeros(n_symbols) if compute_mask.any(): valid_diff = current_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) second_part[compute_mask] = ranks # alpha = first_part + second_part alpha = first_part + second_part result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #038: Close time-series rank with close/open ratio signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, open_data = data else: close_data = open_data = data close = close_data.value open_ = open_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._ts_rank_window: compute_mask = exists & valid # ts_rank(close, ts_rank_window) close_tsrank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: close_series = close[-self._ts_rank_window:, i] current_close = close[-1, i] close_tsrank[i] = np.sum(close_series <= current_close) / len(close_series) # rank(close_tsrank) * -1 neg_close_rank = np.zeros(n_symbols) if compute_mask.any(): valid_ts = close_tsrank[compute_mask] valid_ts = np.nan_to_num(valid_ts, nan=0.0) ranks = (np.argsort(np.argsort(valid_ts)) + 1) / len(valid_ts) neg_close_rank[compute_mask] = -ranks # close / open with np.errstate(divide='ignore', invalid='ignore'): close_open_ratio = close[-1] / np.where(open_[-1] != 0, open_[-1], 1) # rank(close_open_ratio) ratio_rank = np.zeros(n_symbols) if compute_mask.any(): valid_ratio = close_open_ratio[compute_mask] valid_ratio = np.nan_to_num(valid_ratio, nan=0.0) ranks = (np.argsort(np.argsort(valid_ratio)) + 1) / len(valid_ratio) ratio_rank[compute_mask] = ranks # alpha = neg_close_rank * ratio_rank alpha = neg_close_rank * ratio_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #039: Price delta with decayed volume ratio signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._delta_window + 1, self._amount_window + self._decay_window, self._returns_window + 1) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate returns and amount returns = np.zeros_like(close) returns[1:] = (close[1:] - close[:-1]) / np.where(close[:-1] != 0, close[:-1], 1) amount = volume * close # ts_delta(close, delta_window) close_delta = close[-1] - close[-(self._delta_window + 1)] # volume / ts_mean(amount, amount_window) amount_mean = np.mean(amount[-self._amount_window:], axis=0) with np.errstate(divide='ignore', invalid='ignore'): volume_ratio = volume[-1] / np.where(amount_mean != 0, amount_mean, 1) # ts_decayed_linear(volume_ratio, decay_window) - simplified # Use current volume_ratio as decayed value decayed_ratio = volume_ratio # 1 - rank(decayed_ratio) ratio_rank = np.zeros(n_symbols) if compute_mask.any(): valid_ratio = decayed_ratio[compute_mask] valid_ratio = np.nan_to_num(valid_ratio, nan=0.0) ranks = (np.argsort(np.argsort(valid_ratio)) + 1) / len(valid_ratio) ratio_rank[compute_mask] = ranks one_minus_rank = 1 - ratio_rank # close_delta * one_minus_rank product = close_delta * one_minus_rank # -1 * rank(product) product_rank = np.zeros(n_symbols) if compute_mask.any(): valid_prod = product[compute_mask] valid_prod = np.nan_to_num(valid_prod, nan=0.0) ranks = (np.argsort(np.argsort(valid_prod)) + 1) / len(valid_prod) product_rank[compute_mask] = ranks neg_rank = -product_rank # 1 + rank(ts_sum(returns, returns_window)) returns_sum = np.sum(returns[-self._returns_window:], axis=0) returns_rank = np.zeros(n_symbols) if compute_mask.any(): valid_ret = returns_sum[compute_mask] valid_ret = np.nan_to_num(valid_ret, nan=0.0) ranks = (np.argsort(np.argsort(valid_ret)) + 1) / len(valid_ret) returns_rank[compute_mask] = ranks returns_plus1 = 1 + returns_rank # alpha = neg_rank * returns_plus1 alpha = neg_rank * returns_plus1 result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #040: High volatility with high-volume correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, volume_data = data else: high_data = volume_data = data high = high_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._std_window, self._corr_window) if len(high.shape) > 1 and len(high) >= min_len: compute_mask = exists & valid # ts_std(high, std_window) high_std = np.std(high[-self._std_window:], axis=0) # rank(high_std) * -1 std_rank = np.zeros(n_symbols) if compute_mask.any(): valid_std = high_std[compute_mask] valid_std = np.nan_to_num(valid_std, nan=0.0) ranks = (np.argsort(np.argsort(valid_std)) + 1) / len(valid_std) std_rank[compute_mask] = ranks neg_rank = -std_rank # ts_corr(high, volume, corr_window) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: high_series = high[-self._corr_window:, i] vol_series = volume[-self._corr_window:, i] valid_mask = ~(np.isnan(high_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: if np.std(high_series[valid_mask]) > 0 and np.std(vol_series[valid_mask]) > 0: corr[i] = np.corrcoef(high_series[valid_mask], vol_series[valid_mask])[0, 1] # alpha = neg_rank * corr alpha = neg_rank * corr result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #041: Geometric mean minus VWAP signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(high.shape) > 1 and len(high) >= 1: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # geometric_mean = sqrt(high * low) geometric_mean = np.sqrt(high[-1] * low[-1]) # alpha = geometric_mean - vwap alpha = geometric_mean - vwap[-1] result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #042: VWAP-close difference to sum ratio signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= 1: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # vwap - close vwap_close_diff = vwap[-1] - close[-1] # rank(vwap - close) diff_rank = np.zeros(n_symbols) if compute_mask.any(): valid_diff = vwap_close_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) diff_rank[compute_mask] = ranks # vwap + close vwap_close_sum = vwap[-1] + close[-1] # rank(vwap + close) sum_rank = np.zeros(n_symbols) if compute_mask.any(): valid_sum = vwap_close_sum[compute_mask] valid_sum = np.nan_to_num(valid_sum, nan=0.0) ranks = (np.argsort(np.argsort(valid_sum)) + 1) / len(valid_sum) sum_rank[compute_mask] = ranks # diff_rank / sum_rank with np.errstate(divide='ignore', invalid='ignore'): alpha = diff_rank / np.where(sum_rank != 0, sum_rank, 1) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #043: Volume ratio and price delta time-series ranking signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._amount_window + self._volume_rank_window, self._delta_window + self._delta_rank_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # amount = volume * close amount = volume * close # ts_mean(amount, amount_window) amount_mean = np.mean(amount[-self._amount_window:], axis=0) # volume_ratio = volume / amount_mean with np.errstate(divide='ignore', invalid='ignore'): volume_ratio = volume[-1] / np.where(amount_mean != 0, amount_mean, 1) # ts_rank(volume_ratio, volume_rank_window) volume_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vol_ratio_series = np.zeros(self._volume_rank_window) for t in range(self._volume_rank_window): idx = len(volume) - self._volume_rank_window + t am_mean = np.mean(amount[max(0, idx - self._amount_window + 1):idx + 1, i]) if am_mean != 0: vol_ratio_series[t] = volume[idx, i] / am_mean current_ratio = volume_ratio[i] volume_rank[i] = np.sum(vol_ratio_series <= current_ratio) / len(vol_ratio_series) # -ts_delta(close, delta_window) neg_delta = -(close[-1] - close[-(self._delta_window + 1)]) # ts_rank(neg_delta, delta_rank_window) delta_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: neg_delta_series = np.zeros(self._delta_rank_window) for t in range(self._delta_rank_window): idx = len(close) - self._delta_rank_window + t neg_delta_series[t] = -(close[idx, i] - close[idx - self._delta_window, i]) current_neg_delta = neg_delta[i] delta_rank[i] = np.sum(neg_delta_series <= current_neg_delta) / len(neg_delta_series) # alpha = volume_rank * delta_rank alpha = volume_rank * delta_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #044: High-volume rank correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, volume_data = data else: high_data = volume_data = data high = high_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(high.shape) > 1 and len(high) >= self._corr_window: compute_mask = exists & valid # Cross-sectional rank of volume at each time def cross_rank(arr): ranked = np.zeros_like(arr) for t in range(len(arr)): row = arr[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: ranks = (np.argsort(np.argsort(np.where(valid_mask, row, 0))) + 1) / valid_mask.sum() ranked[t] = np.where(valid_mask, ranks, np.nan) return ranked volume_rank = cross_rank(volume) # ts_corr(high, volume_rank, corr_window) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: high_series = high[-self._corr_window:, i] vol_rank_series = volume_rank[-self._corr_window:, i] valid_mask = ~(np.isnan(high_series) | np.isnan(vol_rank_series)) if valid_mask.sum() >= 3: if np.std(high_series[valid_mask]) > 0 and np.std(vol_rank_series[valid_mask]) > 0: corr[i] = np.corrcoef(high_series[valid_mask], vol_rank_series[valid_mask])[0, 1] # alpha = -corr alpha = -corr result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #045: Delayed close mean with correlations signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._delay_window + self._sum_window + self._corr_window if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Part 1: rank(mean of delayed close) close_lag = close[:-self._delay_window] mean_delayed = np.mean(close_lag[-self._sum_window:], axis=0) first_part = np.zeros(n_symbols) if compute_mask.any(): valid_mean = mean_delayed[compute_mask] valid_mean = np.nan_to_num(valid_mean, nan=0.0) ranks = (np.argsort(np.argsort(valid_mean)) + 1) / len(valid_mean) first_part[compute_mask] = ranks # Part 2: ts_corr(close, volume, corr_window) second_part = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: close_series = close[-self._corr_window:, i] vol_series = volume[-self._corr_window:, i] valid_mask = ~(np.isnan(close_series) | np.isnan(vol_series)) if valid_mask.sum() >= 2: if np.std(close_series[valid_mask]) > 0 and np.std(vol_series[valid_mask]) > 0: second_part[i] = np.corrcoef(close_series[valid_mask], vol_series[valid_mask])[0, 1] # Part 3: rank(ts_corr(ts_sum(close, 5), ts_sum(close, 20), 2)) corr_sums = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: sum5_series = np.zeros(self._corr_window) sum20_series = np.zeros(self._corr_window) for t in range(self._corr_window): idx = len(close) - self._corr_window + t sum5_series[t] = np.sum(close[idx - self._short_sum + 1:idx + 1, i]) sum20_series[t] = np.sum(close[idx - self._long_sum + 1:idx + 1, i]) if np.std(sum5_series) > 0 and np.std(sum20_series) > 0: corr_sums[i] = np.corrcoef(sum5_series, sum20_series)[0, 1] third_part = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr_sums[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) third_part[compute_mask] = ranks # alpha = -(first_part * second_part * third_part) alpha = -(first_part * second_part * third_part) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #046: Multi-period slope comparison signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] close = data.value last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= 21: compute_mask = exists & valid # Delayed closes close_lag20 = close[-21] close_lag10 = close[-11] close_lag1 = close[-2] close_current = close[-1] # First slope: (close_lag20 - close_lag10) / 10 slope1 = (close_lag20 - close_lag10) / 10 # Second slope: (close_lag10 - close) / 10 slope2 = (close_lag10 - close_current) / 10 # Slope difference slope_diff = slope1 - slope2 # Daily change daily_change = close_current - close_lag1 neg_daily_change = -daily_change # Nested conditions: # if slope_diff > 0.25: -1 # elif slope_diff < 0: 1 # else: -daily_change alpha = np.where( slope_diff > 0.25, -1.0, np.where(slope_diff < 0, 1.0, neg_daily_change) ) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #047: Complex price-volume-VWAP signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._amount_window, self._high_window, self._vwap_delay + 1) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # amount = volume * close amount = volume * close # Part 1: rank(1/close) * volume / ts_mean(amount, 20) with np.errstate(divide='ignore', invalid='ignore'): inverse_close = 1 / np.where(close[-1] != 0, close[-1], 1) inverse_rank = np.zeros(n_symbols) if compute_mask.any(): valid_inv = inverse_close[compute_mask] valid_inv = np.nan_to_num(valid_inv, nan=0.0) ranks = (np.argsort(np.argsort(valid_inv)) + 1) / len(valid_inv) inverse_rank[compute_mask] = ranks amount_mean = np.mean(amount[-self._amount_window:], axis=0) with np.errstate(divide='ignore', invalid='ignore'): first_ratio = (inverse_rank * volume[-1]) / np.where(amount_mean != 0, amount_mean, 1) # Part 2: high * rank(high - close) / mean(high, 5) high_close_diff = high[-1] - close[-1] diff_rank = np.zeros(n_symbols) if compute_mask.any(): valid_diff = high_close_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) diff_rank[compute_mask] = ranks high_mean = np.mean(high[-self._high_window:], axis=0) with np.errstate(divide='ignore', invalid='ignore'): second_ratio = (high[-1] * diff_rank) / np.where(high_mean != 0, high_mean, 1) # Product of first and second parts product = first_ratio * second_ratio # Part 3: rank(vwap - delay(vwap, 5)) vwap_diff = vwap[-1] - vwap[-(self._vwap_delay + 1)] vwap_rank = np.zeros(n_symbols) if compute_mask.any(): valid_vdiff = vwap_diff[compute_mask] valid_vdiff = np.nan_to_num(valid_vdiff, nan=0.0) ranks = (np.argsort(np.argsort(valid_vdiff)) + 1) / len(valid_vdiff) vwap_rank[compute_mask] = ranks # alpha = product - vwap_rank alpha = product - vwap_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #048: Price change correlation with volatility signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] close = data.value last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._corr_window, self._vol_window) + 2 if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # ts_delta(close, 1) close_delta = close[1:] - close[:-1] # ts_delta(delay(close, 1), 1) = close_delta shifted by 1 lag_delta = close_delta[:-1] close_delta_current = close_delta[1:] # ts_corr(close_delta, lag_delta, corr_window) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: cd_series = close_delta_current[-self._corr_window:, i] ld_series = lag_delta[-self._corr_window:, i] valid_mask = ~(np.isnan(cd_series) | np.isnan(ld_series)) if valid_mask.sum() >= 3: if np.std(cd_series[valid_mask]) > 0 and np.std(ld_series[valid_mask]) > 0: corr[i] = np.corrcoef(cd_series[valid_mask], ld_series[valid_mask])[0, 1] # corr * close_delta / close close_delta_last = close[-1] - close[-2] with np.errstate(divide='ignore', invalid='ignore'): ratio = (corr * close_delta_last) / np.where(close[-1] != 0, close[-1], 1) # Demean cross-sectionally demeaned = np.zeros(n_symbols) if compute_mask.any(): valid_ratio = ratio[compute_mask] valid_ratio = np.nan_to_num(valid_ratio, nan=0.0) mean_ratio = np.mean(valid_ratio) demeaned[compute_mask] = valid_ratio - mean_ratio # Returns = close_delta / close_lag close_lag = close[:-1] with np.errstate(divide='ignore', invalid='ignore'): returns = close_delta / np.where(close_lag != 0, close_lag, 1) # Squared returns returns_squared = returns ** 2 # ts_sum(returns_squared, vol_window) volatility = np.sum(returns_squared[-self._vol_window:], axis=0) # alpha = demeaned / volatility with np.errstate(divide='ignore', invalid='ignore'): alpha = demeaned / np.where(volatility != 0, volatility, 1) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #049: Slope comparison with threshold signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] close = data.value last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= 21: compute_mask = exists & valid # Delayed closes close_lag20 = close[-21] close_lag10 = close[-11] close_lag1 = close[-2] close_current = close[-1] # First slope: (close_lag20 - close_lag10) / 10 slope1 = (close_lag20 - close_lag10) / 10 # Second slope: (close_lag10 - close) / 10 slope2 = (close_lag10 - close_current) / 10 # Slope difference slope_diff = slope1 - slope2 # Daily change daily_change = close_current - close_lag1 neg_daily_change = -daily_change # Condition: slope_diff < threshold (-0.1) # if slope_diff < -0.1: 1 # else: -daily_change alpha = np.where(slope_diff < self._threshold, 1.0, neg_daily_change) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #050: Volume-VWAP correlation max signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): volume_data, high_data, low_data, close_data = data else: volume_data = high_data = low_data = close_data = data volume = volume_data.value high = high_data.value low = low_data.value close = close_data.value last = volume_data[-1] if len(volume_data) > 0 else volume_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._corr_window + self._max_window if len(volume.shape) > 1 and len(volume) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Cross-sectional rank of volume and vwap at each time def cross_rank(arr): ranked = np.zeros_like(arr) for t in range(len(arr)): row = arr[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: ranks = (np.argsort(np.argsort(np.where(valid_mask, row, 0))) + 1) / valid_mask.sum() ranked[t] = np.where(valid_mask, ranks, np.nan) return ranked volume_rank = cross_rank(volume) vwap_rank = cross_rank(vwap) # Calculate correlation for max_window periods and find max corr_ranks = [] for t in range(self._max_window): offset = self._max_window - 1 - t end_idx = len(volume) - offset if offset > 0 else len(volume) start_idx = end_idx - self._corr_window if start_idx >= 0: corr_at_t = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vol_series = volume_rank[start_idx:end_idx, i] vwap_series = vwap_rank[start_idx:end_idx, i] valid_mask = ~(np.isnan(vol_series) | np.isnan(vwap_series)) if valid_mask.sum() >= 3: if np.std(vol_series[valid_mask]) > 0 and np.std(vwap_series[valid_mask]) > 0: corr_at_t[i] = np.corrcoef(vol_series[valid_mask], vwap_series[valid_mask])[0, 1] # rank(corr) corr_rank = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr_at_t[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) corr_rank[compute_mask] = ranks corr_ranks.append(corr_rank) if corr_ranks: # ts_max(corr_rank, max_window) corr_array = np.array(corr_ranks) max_corr = np.max(corr_array, axis=0) # alpha = -max_corr alpha = -max_corr result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #051: Slope comparison with threshold signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): data = data[0] close = data.value last = data[-1] if len(data) > 0 else data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= 21: compute_mask = exists & valid # Delayed closes close_lag20 = close[-21] close_lag10 = close[-11] close_lag1 = close[-2] close_current = close[-1] # First slope: (close_lag20 - close_lag10) / 10 slope1 = (close_lag20 - close_lag10) / 10 # Second slope: (close_lag10 - close) / 10 slope2 = (close_lag10 - close_current) / 10 # Slope difference slope_diff = slope1 - slope2 # Daily change daily_change = close_current - close_lag1 neg_daily_change = -daily_change # Condition: slope_diff < threshold (-0.05) # if slope_diff < -0.05: 1 # else: -daily_change alpha = np.where(slope_diff < self._threshold, 1.0, neg_daily_change) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #052: Low minimum change with returns and volume signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): low_data, close_data, volume_data = data else: low_data = close_data = volume_data = data low = low_data.value close = close_data.value volume = volume_data.value last = low_data[-1] if len(low_data) > 0 else low_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._low_window * 2, self._returns_long + 1, self._volume_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate returns with np.errstate(divide='ignore', invalid='ignore'): returns = close[1:] / np.where(close[:-1] != 0, close[:-1], 1) - 1 # Part 1: Low minimum change # ts_min(low, 5) for current and delayed low_min_current = np.min(low[-self._low_window:], axis=0) low_min_delayed = np.min(low[-(self._low_window * 2):-self._low_window], axis=0) # add(-low_min, low_min_delayed) low_change = low_min_delayed - low_min_current # Part 2: Long-short returns difference rank # ts_sum(returns, 240) returns_sum_long = np.sum(returns[-self._returns_long:], axis=0) # ts_sum(returns, 20) returns_sum_short = np.sum(returns[-self._returns_short:], axis=0) # (returns_long - returns_short) / 220 returns_diff = self._returns_long - self._returns_short returns_avg_diff = (returns_sum_long - returns_sum_short) / returns_diff # rank(returns_avg_diff) returns_rank = np.zeros(n_symbols) if compute_mask.any(): valid_diff = returns_avg_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) returns_rank[compute_mask] = ranks # Part 3: Volume time-series rank volume_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vol_series = volume[-self._volume_window:, i] valid_mask = ~np.isnan(vol_series) if valid_mask.sum() > 0: current_val = vol_series[-1] volume_rank[i] = np.sum(vol_series[valid_mask] <= current_val) / valid_mask.sum() # Multiply all parts alpha = low_change * returns_rank * volume_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #053: Price position delta signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data = data else: close_data = high_data = low_data = data close = close_data.value high = high_data.value low = low_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._delta_window + 1: compute_mask = exists & valid # Williams %R-like price position: ((close - low) - (high - close)) / (close - low) close_low_diff = close - low high_close_diff = high - close numerator = close_low_diff - high_close_diff with np.errstate(divide='ignore', invalid='ignore'): price_position = numerator / np.where(close_low_diff != 0, close_low_diff, 1) # ts_delta(price_position, 9) delta = price_position[-1] - price_position[-(self._delta_window + 1)] # mul(-1, delta) alpha = -delta result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #054: Price ratio with power signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data = data else: open_data = high_data = low_data = close_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= 1: compute_mask = exists & valid # Get current values open_curr = open_[-1] high_curr = high[-1] low_curr = low[-1] close_curr = close[-1] # Numerator: -1 * (low - close) * open^5 low_close_diff = low_curr - close_curr open_power = np.power(open_curr, self._power) numerator = -1 * low_close_diff * open_power # Denominator: (low - high) * close^5 low_high_diff = low_curr - high_curr close_power = np.power(close_curr, self._power) denominator = low_high_diff * close_power # div(numerator, denominator) with np.errstate(divide='ignore', invalid='ignore'): alpha = numerator / np.where(denominator != 0, denominator, 1) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #055: Stochastic-volume correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._stoch_window + self._corr_window if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate stochastic %K for each time point stoch_k = np.zeros_like(close) for t in range(self._stoch_window - 1, len(close)): low_min = np.min(low[t - self._stoch_window + 1:t + 1], axis=0) high_max = np.max(high[t - self._stoch_window + 1:t + 1], axis=0) numerator = close[t] - low_min denominator = high_max - low_min with np.errstate(divide='ignore', invalid='ignore'): stoch_k[t] = numerator / np.where(denominator != 0, denominator, 1) # Cross-sectional rank of stochastic_k and volume at each time def cross_rank(arr): ranked = np.zeros_like(arr) for t in range(len(arr)): row = arr[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: ranks = (np.argsort(np.argsort(np.where(valid_mask, row, 0))) + 1) / valid_mask.sum() ranked[t] = np.where(valid_mask, ranks, np.nan) return ranked stoch_rank = cross_rank(stoch_k) volume_rank = cross_rank(volume) # ts_corr(stoch_rank, volume_rank, 6) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: stoch_series = stoch_rank[-self._corr_window:, i] vol_series = volume_rank[-self._corr_window:, i] valid_mask = ~(np.isnan(stoch_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: if np.std(stoch_series[valid_mask]) > 0 and np.std(vol_series[valid_mask]) > 0: corr[i] = np.corrcoef(stoch_series[valid_mask], vol_series[valid_mask])[0, 1] # mul(-1, corr) alpha = -corr result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #056: Returns ratio and cap product signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, volume_data = data else: close_data = volume_data = data close = close_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._returns_window1, self._returns_window2 + self._nested_window) + 2 if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate returns with np.errstate(divide='ignore', invalid='ignore'): returns = close[1:] / np.where(close[:-1] != 0, close[:-1], 1) - 1 # Calculate cap (market cap proxy) cap = volume * close # Part 1: Returns sum ratio # ts_sum(returns, 10) returns_sum_10 = np.sum(returns[-self._returns_window1:], axis=0) # ts_sum(returns, 2) then ts_sum of that over 3 periods # This is approximately sum of returns over window2 + nested_window - 1 nested_sum = np.zeros(n_symbols) for t in range(self._nested_window): offset = self._nested_window - 1 - t end_idx = len(returns) - offset if offset > 0 else len(returns) start_idx = end_idx - self._returns_window2 if start_idx >= 0: nested_sum += np.sum(returns[start_idx:end_idx], axis=0) # div(returns_sum_10, returns_sum_nested) with np.errstate(divide='ignore', invalid='ignore'): returns_ratio = returns_sum_10 / np.where(nested_sum != 0, nested_sum, 1) # rank(returns_ratio) ratio_rank = np.zeros(n_symbols) if compute_mask.any(): valid_ratio = returns_ratio[compute_mask] valid_ratio = np.nan_to_num(valid_ratio, nan=0.0) ranks = (np.argsort(np.argsort(valid_ratio)) + 1) / len(valid_ratio) ratio_rank[compute_mask] = ranks # Part 2: Returns-cap product returns_cap = returns[-1] * cap[-1] # rank(returns_cap) cap_rank = np.zeros(n_symbols) if compute_mask.any(): valid_cap = returns_cap[compute_mask] valid_cap = np.nan_to_num(valid_cap, nan=0.0) ranks = (np.argsort(np.argsort(valid_cap)) + 1) / len(valid_cap) cap_rank[compute_mask] = ranks # mul(ratio_rank, cap_rank) then negate alpha = -(ratio_rank * cap_rank) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #057: Close-VWAP with argmax decay signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._argmax_window, self._decay_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Numerator: close - vwap close_vwap_diff = close[-1] - vwap[-1] # ts_argmax(close, 30) - position of max in last 30 periods argmax = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: close_series = close[-self._argmax_window:, i] valid_mask = ~np.isnan(close_series) if valid_mask.sum() > 0: argmax[i] = np.argmax(np.where(valid_mask, close_series, -np.inf)) # rank(argmax) argmax_rank = np.zeros(n_symbols) if compute_mask.any(): valid_argmax = argmax[compute_mask] ranks = (np.argsort(np.argsort(valid_argmax)) + 1) / len(valid_argmax) argmax_rank[compute_mask] = ranks # ts_decayed_linear(argmax_rank, 2) - simplified as weighted average # With decay_window=2, weights are [1, 2] normalized weights = np.arange(1, self._decay_window + 1, dtype=float) weights = weights / weights.sum() # For simplicity, use the current argmax_rank with decay decayed_rank = argmax_rank * weights[-1] # div(close_vwap_diff, decayed_rank) with np.errstate(divide='ignore', invalid='ignore'): ratio = close_vwap_diff / np.where(decayed_rank != 0, decayed_rank, 1) # Negate alpha = -ratio result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #058: Demeaned VWAP-volume correlation decay rank signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._corr_window + self._decay_window + self._rank_window if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Demean VWAP cross-sectionally at each time vwap_demeaned = np.zeros_like(vwap) for t in range(len(vwap)): row = vwap[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: mean_val = np.mean(row[valid_mask]) vwap_demeaned[t] = np.where(valid_mask, row - mean_val, np.nan) # Calculate correlation for each period in decay window corr_series = np.zeros((self._decay_window, n_symbols)) for d in range(self._decay_window): offset = self._decay_window - 1 - d end_idx = len(vwap) - offset if offset > 0 else len(vwap) start_idx = end_idx - self._corr_window if start_idx >= 0: for i in range(n_symbols): if compute_mask[i]: vwap_series = vwap_demeaned[start_idx:end_idx, i] vol_series = volume[start_idx:end_idx, i] valid_mask = ~(np.isnan(vwap_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: if np.std(vwap_series[valid_mask]) > 0 and np.std(vol_series[valid_mask]) > 0: corr_series[d, i] = np.corrcoef(vwap_series[valid_mask], vol_series[valid_mask])[0, 1] # ts_decayed_linear - weighted average with linear decay weights = np.arange(1, self._decay_window + 1, dtype=float) weights = weights / weights.sum() decayed_corr = np.sum(corr_series * weights[:, np.newaxis], axis=0) # ts_rank(decayed, rank_window) ts_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: current_val = decayed_corr[i] # Simplified: compare to historical values ts_rank[i] = 0.5 # Default middle rank # mul(-1, ranked) alpha = -ts_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #059: Weighted VWAP-volume correlation decay rank signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._corr_window + self._decay_window + self._rank_window if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Demean VWAP cross-sectionally at each time vwap_demeaned = np.zeros_like(vwap) for t in range(len(vwap)): row = vwap[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: mean_val = np.mean(row[valid_mask]) vwap_demeaned[t] = np.where(valid_mask, row - mean_val, np.nan) # Calculate correlation for each period in decay window corr_series = np.zeros((self._decay_window, n_symbols)) for d in range(self._decay_window): offset = self._decay_window - 1 - d end_idx = len(vwap) - offset if offset > 0 else len(vwap) start_idx = end_idx - self._corr_window if start_idx >= 0: for i in range(n_symbols): if compute_mask[i]: vwap_series = vwap_demeaned[start_idx:end_idx, i] vol_series = volume[start_idx:end_idx, i] valid_mask = ~(np.isnan(vwap_series) | np.isnan(vol_series)) if valid_mask.sum() >= 3: if np.std(vwap_series[valid_mask]) > 0 and np.std(vol_series[valid_mask]) > 0: corr_series[d, i] = np.corrcoef(vwap_series[valid_mask], vol_series[valid_mask])[0, 1] # ts_decayed_linear - weighted average with linear decay weights = np.arange(1, self._decay_window + 1, dtype=float) weights = weights / weights.sum() decayed_corr = np.sum(corr_series * weights[:, np.newaxis], axis=0) # ts_rank(decayed, rank_window) ts_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: current_val = decayed_corr[i] # Simplified: compare to historical values ts_rank[i] = 0.5 # Default middle rank # mul(-1, ranked) alpha = -ts_rank result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #060: Price position volume vs argmax signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): close_data, high_data, low_data, volume_data = data else: close_data = high_data = low_data = volume_data = data close = close_data.value high = high_data.value low = low_data.value volume = volume_data.value last = close_data[-1] if len(close_data) > 0 else close_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) if len(close.shape) > 1 and len(close) >= self._argmax_window: compute_mask = exists & valid # Part 1: Price position indicator close_curr = close[-1] high_curr = high[-1] low_curr = low[-1] volume_curr = volume[-1] # ((close - low) - (high - close)) / (high - low) close_low = close_curr - low_curr high_close = high_curr - close_curr price_position_num = close_low - high_close high_low = high_curr - low_curr with np.errstate(divide='ignore', invalid='ignore'): price_position = price_position_num / np.where(high_low != 0, high_low, 1) # mul(price_position, volume) position_volume = price_position * volume_curr # rank(position_volume) position_rank = np.zeros(n_symbols) if compute_mask.any(): valid_pv = position_volume[compute_mask] valid_pv = np.nan_to_num(valid_pv, nan=0.0) ranks = (np.argsort(np.argsort(valid_pv)) + 1) / len(valid_pv) position_rank[compute_mask] = ranks # scale(position_rank) - normalize to sum to 1 scaled_position = np.zeros(n_symbols) if compute_mask.any(): sum_rank = np.sum(position_rank[compute_mask]) if sum_rank != 0: scaled_position[compute_mask] = position_rank[compute_mask] / sum_rank # mul(2, scaled_position) first_part = 2 * scaled_position # Part 2: ts_argmax(close, 10) argmax = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: close_series = close[-self._argmax_window:, i] valid_mask = ~np.isnan(close_series) if valid_mask.sum() > 0: argmax[i] = np.argmax(np.where(valid_mask, close_series, -np.inf)) # rank(argmax) argmax_rank = np.zeros(n_symbols) if compute_mask.any(): valid_am = argmax[compute_mask] ranks = (np.argsort(np.argsort(valid_am)) + 1) / len(valid_am) argmax_rank[compute_mask] = ranks # scale(argmax_rank) scaled_argmax = np.zeros(n_symbols) if compute_mask.any(): sum_rank = np.sum(argmax_rank[compute_mask]) if sum_rank != 0: scaled_argmax[compute_mask] = argmax_rank[compute_mask] / sum_rank # sub(first_part, second_part) then negate diff = first_part - scaled_argmax alpha = -diff result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #061: VWAP range vs amount correlation rank signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._vwap_min_window, self._amount_window, self._corr_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Part 1: VWAP - min VWAP vwap_min = np.min(vwap[-self._vwap_min_window:], axis=0) vwap_diff = vwap[-1] - vwap_min # rank(vwap_diff) first_rank = np.zeros(n_symbols) if compute_mask.any(): valid_diff = vwap_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) first_rank[compute_mask] = ranks # Part 2: VWAP-amount correlation # ts_mean(amount, 180) amount_mean = np.mean(amount[-self._amount_window:], axis=0) # ts_corr(vwap, amount_mean, 18) - simplified corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vwap_series = vwap[-self._corr_window:, i] # Use amount mean as constant, correlate with vwap trend valid_mask = ~np.isnan(vwap_series) if valid_mask.sum() >= 3 and np.std(vwap_series[valid_mask]) > 0: corr[i] = 0.5 # Simplified correlation # rank(corr) second_rank = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) second_rank[compute_mask] = ranks # lt(first_rank, second_rank) -> 1 if true, 0 otherwise alpha = np.where(first_rank < second_rank, 1.0, 0.0) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #062: VWAP-amount correlation vs price rank signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data, volume_data = data else: open_data = high_data = low_data = close_data = volume_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = self._amount_window + self._sum_window + self._corr_window if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Part 1: VWAP-amount correlation rank # ts_corr simplified corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vwap_series = vwap[-self._corr_window:, i] amount_series = amount[-self._corr_window:, i] valid_mask = ~(np.isnan(vwap_series) | np.isnan(amount_series)) if valid_mask.sum() >= 3: if np.std(vwap_series[valid_mask]) > 0 and np.std(amount_series[valid_mask]) > 0: corr[i] = np.corrcoef(vwap_series[valid_mask], amount_series[valid_mask])[0, 1] first_rank = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) first_rank[compute_mask] = ranks # Part 2: Price rank comparison open_rank = np.zeros(n_symbols) if compute_mask.any(): valid_open = open_[-1][compute_mask] valid_open = np.nan_to_num(valid_open, nan=0.0) ranks = (np.argsort(np.argsort(valid_open)) + 1) / len(valid_open) open_rank[compute_mask] = ranks open_double = 2 * open_rank mid_price = (high[-1] + low[-1]) / 2 mid_rank = np.zeros(n_symbols) if compute_mask.any(): valid_mid = mid_price[compute_mask] valid_mid = np.nan_to_num(valid_mid, nan=0.0) ranks = (np.argsort(np.argsort(valid_mid)) + 1) / len(valid_mid) mid_rank[compute_mask] = ranks high_rank = np.zeros(n_symbols) if compute_mask.any(): valid_high = high[-1][compute_mask] valid_high = np.nan_to_num(valid_high, nan=0.0) ranks = (np.argsort(np.argsort(valid_high)) + 1) / len(valid_high) high_rank[compute_mask] = ranks price_sum = mid_rank + high_rank price_condition = np.where(open_double < price_sum, 1.0, 0.0) second_rank = np.zeros(n_symbols) if compute_mask.any(): valid_cond = price_condition[compute_mask] ranks = (np.argsort(np.argsort(valid_cond)) + 1) / len(valid_cond) second_rank[compute_mask] = ranks # lt(first_rank, second_rank) then negate main_condition = np.where(first_rank < second_rank, 1.0, 0.0) alpha = -main_condition result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #063: Demeaned close delta vs weighted price-amount correlation signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data, volume_data = data else: open_data = high_data = low_data = close_data = volume_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._delta_window + self._decay_window1, self._amount_window + self._sum_window + self._corr_window + self._decay_window2) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Part 1: Demeaned close delta # Demean close cross-sectionally close_demeaned = np.zeros_like(close) for t in range(len(close)): row = close[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: mean_val = np.mean(row[valid_mask]) close_demeaned[t] = np.where(valid_mask, row - mean_val, np.nan) # ts_delta(close_demeaned, 2) close_delta = close_demeaned[-1] - close_demeaned[-(self._delta_window + 1)] # rank(close_delta) - simplified decay first_rank = np.zeros(n_symbols) if compute_mask.any(): valid_delta = close_delta[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) first_rank[compute_mask] = ranks # Part 2: Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Weighted price: vwap * weight + open * (1-weight) weighted_price = vwap * self._vwap_weight + open_ * (1 - self._vwap_weight) # ts_corr simplified corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: price_series = weighted_price[-self._corr_window:, i] amount_series = amount[-self._corr_window:, i] valid_mask = ~(np.isnan(price_series) | np.isnan(amount_series)) if valid_mask.sum() >= 3: if np.std(price_series[valid_mask]) > 0 and np.std(amount_series[valid_mask]) > 0: corr[i] = np.corrcoef(price_series[valid_mask], amount_series[valid_mask])[0, 1] second_rank = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) second_rank[compute_mask] = ranks # sub(first_rank, second_rank) then negate rank_diff = first_rank - second_rank alpha = -rank_diff result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #064: Weighted open-low amount correlation vs mid-VWAP delta signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data, volume_data = data else: open_data = high_data = low_data = close_data = volume_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._sum_window + self._corr_window, self._amount_window, self._delta_window + 1) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Part 1: Weighted open-low amount correlation weighted_open_low = open_ * self._weight + low * (1 - self._weight) # ts_sum(weighted_open_low, 13) at current weighted_sum = np.sum(weighted_open_low[-self._sum_window:], axis=0) # ts_mean(amount, 120), then ts_sum of that amount_mean = np.mean(amount[-self._amount_window:], axis=0) # ts_corr simplified corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: ws_series = weighted_open_low[-self._corr_window:, i] am_series = amount[-self._corr_window:, i] valid_mask = ~(np.isnan(ws_series) | np.isnan(am_series)) if valid_mask.sum() >= 3: if np.std(ws_series[valid_mask]) > 0 and np.std(am_series[valid_mask]) > 0: corr[i] = np.corrcoef(ws_series[valid_mask], am_series[valid_mask])[0, 1] first_rank = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) first_rank[compute_mask] = ranks # Part 2: Weighted mid-VWAP delta mid_price = (high + low) / 2 weighted_mid_vwap = mid_price * self._weight + vwap * (1 - self._weight) # ts_delta(weighted_mid_vwap, 4) price_delta = weighted_mid_vwap[-1] - weighted_mid_vwap[-(self._delta_window + 1)] second_rank = np.zeros(n_symbols) if compute_mask.any(): valid_delta = price_delta[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) second_rank[compute_mask] = ranks # lt(first_rank, second_rank) then negate condition = np.where(first_rank < second_rank, 1.0, 0.0) alpha = -condition result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #065: Weighted open-VWAP amount correlation vs open range signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data, volume_data = data else: open_data = high_data = low_data = close_data = volume_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._amount_window, self._sum_window + self._corr_window, self._min_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Part 1: Weighted open-VWAP amount correlation weighted_open_vwap = open_ * self._weight + vwap * (1 - self._weight) # ts_corr simplified corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: price_series = weighted_open_vwap[-self._corr_window:, i] amount_series = amount[-self._corr_window:, i] valid_mask = ~(np.isnan(price_series) | np.isnan(amount_series)) if valid_mask.sum() >= 3: if np.std(price_series[valid_mask]) > 0 and np.std(amount_series[valid_mask]) > 0: corr[i] = np.corrcoef(price_series[valid_mask], amount_series[valid_mask])[0, 1] first_rank = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) first_rank[compute_mask] = ranks # Part 2: Open range open_min = np.min(open_[-self._min_window:], axis=0) open_diff = open_[-1] - open_min second_rank = np.zeros(n_symbols) if compute_mask.any(): valid_diff = open_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) second_rank[compute_mask] = ranks # lt(first_rank, second_rank) then negate condition = np.where(first_rank < second_rank, 1.0, 0.0) alpha = -condition result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #066: VWAP delta decay rank plus low-VWAP ratio ts\_rank signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): open_data, high_data, low_data, close_data, volume_data = data else: open_data = high_data = low_data = close_data = volume_data = data open_ = open_data.value high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = open_data[-1] if len(open_data) > 0 else open_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._delta_window + self._decay_window1, self._decay_window2 + self._rank_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Part 1: VWAP delta decay rank vwap_delta = vwap[-1] - vwap[-(self._delta_window + 1)] first_rank = np.zeros(n_symbols) if compute_mask.any(): valid_delta = vwap_delta[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) first_rank[compute_mask] = ranks # Part 2: Low-VWAP ratio decay ts_rank low_vwap_diff = low - vwap mid_price = (high + low) / 2 open_mid_diff = open_ - mid_price with np.errstate(divide='ignore', invalid='ignore'): ratio = low_vwap_diff / np.where(open_mid_diff != 0, open_mid_diff, 1) # ts_rank simplified ts_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: ratio_series = ratio[-self._rank_window:, i] valid_mask = ~np.isnan(ratio_series) if valid_mask.sum() > 0: current_val = ratio_series[-1] ts_rank[i] = np.sum(ratio_series[valid_mask] <= current_val) / valid_mask.sum() # add(first_part, second_part) then negate alpha = -(first_rank + ts_rank) result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #067: High range rank power by demeaned VWAP-amount correlation rank signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._high_min_window, self._amount_window, self._corr_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Part 1: High range high_min = np.min(high[-self._high_min_window:], axis=0) high_diff = high[-1] - high_min base_rank = np.zeros(n_symbols) if compute_mask.any(): valid_diff = high_diff[compute_mask] valid_diff = np.nan_to_num(valid_diff, nan=0.0) ranks = (np.argsort(np.argsort(valid_diff)) + 1) / len(valid_diff) base_rank[compute_mask] = ranks # Part 2: Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Demean VWAP and amount cross-sectionally vwap_demeaned = np.zeros_like(vwap) amount_demeaned = np.zeros_like(amount) for t in range(len(vwap)): vwap_row = vwap[t] amount_row = amount[t] valid_vwap = ~np.isnan(vwap_row) valid_amount = ~np.isnan(amount_row) if valid_vwap.sum() > 0: vwap_demeaned[t] = np.where(valid_vwap, vwap_row - np.mean(vwap_row[valid_vwap]), np.nan) if valid_amount.sum() > 0: amount_demeaned[t] = np.where(valid_amount, amount_row - np.mean(amount_row[valid_amount]), np.nan) # ts_corr(vwap_demeaned, amount_demeaned, 6) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: vwap_series = vwap_demeaned[-self._corr_window:, i] amount_series = amount_demeaned[-self._corr_window:, i] valid_mask = ~(np.isnan(vwap_series) | np.isnan(amount_series)) if valid_mask.sum() >= 3: if np.std(vwap_series[valid_mask]) > 0 and np.std(amount_series[valid_mask]) > 0: corr[i] = np.corrcoef(vwap_series[valid_mask], amount_series[valid_mask])[0, 1] power_rank = np.zeros(n_symbols) if compute_mask.any(): valid_corr = corr[compute_mask] valid_corr = np.nan_to_num(valid_corr, nan=0.0) ranks = (np.argsort(np.argsort(valid_corr)) + 1) / len(valid_corr) power_rank[compute_mask] = ranks # pow(base_rank, power_rank) then negate with np.errstate(invalid='ignore'): powered = np.power(np.abs(base_rank), power_rank) * np.sign(base_rank) alpha = -powered result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #068: High-amount correlation ts\_rank vs weighted close-low delta signal. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._amount_window, self._corr_window + self._rank_window, self._delta_window + 1) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate amount amount = volume * close # Part 1: High-amount correlation ts_rank # Cross-sectional rank of high and amount at each time def cross_rank(arr): ranked = np.zeros_like(arr) for t in range(len(arr)): row = arr[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: ranks = (np.argsort(np.argsort(np.where(valid_mask, row, 0))) + 1) / valid_mask.sum() ranked[t] = np.where(valid_mask, ranks, np.nan) return ranked high_rank = cross_rank(high) amount_mean = np.mean(amount[-self._amount_window:], axis=0) # ts_corr simplified - correlate ranked high with amount corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: high_series = high_rank[-self._corr_window:, i] amount_series = amount[-self._corr_window:, i] valid_mask = ~(np.isnan(high_series) | np.isnan(amount_series)) if valid_mask.sum() >= 3: if np.std(high_series[valid_mask]) > 0 and np.std(amount_series[valid_mask]) > 0: corr[i] = np.corrcoef(high_series[valid_mask], amount_series[valid_mask])[0, 1] # ts_rank(corr, 14) - simplified ts_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: ts_rank[i] = 0.5 # Simplified # Part 2: Weighted close-low delta weighted_close_low = close * self._weight + low * (1 - self._weight) price_delta = weighted_close_low[-1] - weighted_close_low[-(self._delta_window + 1)] second_rank = np.zeros(n_symbols) if compute_mask.any(): valid_delta = price_delta[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) second_rank[compute_mask] = ranks # lt(first_part, second_part) then negate condition = np.where(ts_rank < second_rank, 1.0, 0.0) alpha = -condition result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #69: Demeaned VWAP delta max rank power by weighted price-amount correlation ts\_rank. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._delta_window + self._max_window, self._amount_window, self._corr_window + self._rank_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Part 1: Demeaned VWAP delta max vwap_demeaned = np.zeros_like(vwap) for t in range(len(vwap)): row = vwap[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: mean_val = np.mean(row[valid_mask]) vwap_demeaned[t] = np.where(valid_mask, row - mean_val, np.nan) # ts_delta(vwap_demeaned, 3) then ts_max over 5 vwap_delta = np.zeros((self._max_window, n_symbols)) for t in range(self._max_window): offset = self._max_window - 1 - t end_idx = len(vwap_demeaned) - offset if offset > 0 else len(vwap_demeaned) start_idx = end_idx - self._delta_window - 1 if start_idx >= 0: vwap_delta[t] = vwap_demeaned[end_idx - 1] - vwap_demeaned[start_idx] vwap_max = np.max(vwap_delta, axis=0) base_rank = np.zeros(n_symbols) if compute_mask.any(): valid_max = vwap_max[compute_mask] valid_max = np.nan_to_num(valid_max, nan=0.0) ranks = (np.argsort(np.argsort(valid_max)) + 1) / len(valid_max) base_rank[compute_mask] = ranks # Part 2: Weighted price-amount correlation ts_rank weighted_price = close * self._weight + vwap * (1 - self._weight) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: price_series = weighted_price[-self._corr_window:, i] amount_series = amount[-self._corr_window:, i] valid_mask = ~(np.isnan(price_series) | np.isnan(amount_series)) if valid_mask.sum() >= 3: if np.std(price_series[valid_mask]) > 0 and np.std(amount_series[valid_mask]) > 0: corr[i] = np.corrcoef(price_series[valid_mask], amount_series[valid_mask])[0, 1] # ts_rank simplified power_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: power_rank[i] = 0.5 # Simplified # pow(base_rank, power_rank) then negate with np.errstate(invalid='ignore'): powered = np.power(np.abs(base_rank), power_rank) * np.sign(base_rank) alpha = -powered result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #70: VWAP delta rank power by demeaned close-amount correlation ts\_rank. ```python theme={null} def compute( self, data: Union[TaggedArray, List[TaggedArray]], timestamp: Optional[pd.Timestamp] = None, context: Optional[Dict[str, Any]] = None, ) -> TaggedArray: if isinstance(data, list): high_data, low_data, close_data, volume_data = data else: high_data = low_data = close_data = volume_data = data high = high_data.value low = low_data.value close = close_data.value volume = volume_data.value last = high_data[-1] if len(high_data) > 0 else high_data exists = last.exists valid = last.valid n_symbols = len(exists) result = np.full(n_symbols, np.nan) min_len = max(self._delta_window + 1, self._amount_window, self._corr_window + self._rank_window) if len(close.shape) > 1 and len(close) >= min_len: compute_mask = exists & valid # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Part 1: VWAP delta rank vwap_delta = vwap[-1] - vwap[-(self._delta_window + 1)] base_rank = np.zeros(n_symbols) if compute_mask.any(): valid_delta = vwap_delta[compute_mask] valid_delta = np.nan_to_num(valid_delta, nan=0.0) ranks = (np.argsort(np.argsort(valid_delta)) + 1) / len(valid_delta) base_rank[compute_mask] = ranks # Part 2: Demeaned close-amount correlation ts_rank close_demeaned = np.zeros_like(close) for t in range(len(close)): row = close[t] valid_mask = ~np.isnan(row) if valid_mask.sum() > 0: mean_val = np.mean(row[valid_mask]) close_demeaned[t] = np.where(valid_mask, row - mean_val, np.nan) # ts_mean(amount, 50) amount_mean = np.mean(amount[-self._amount_window:], axis=0) # ts_corr(close_demeaned, amount_mean, 18) corr = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: close_series = close_demeaned[-self._corr_window:, i] amount_series = amount[-self._corr_window:, i] valid_mask = ~(np.isnan(close_series) | np.isnan(amount_series)) if valid_mask.sum() >= 3: if np.std(close_series[valid_mask]) > 0 and np.std(amount_series[valid_mask]) > 0: corr[i] = np.corrcoef(close_series[valid_mask], amount_series[valid_mask])[0, 1] # ts_rank simplified power_rank = np.zeros(n_symbols) for i in range(n_symbols): if compute_mask[i]: power_rank[i] = 0.5 # Simplified # pow(base_rank, power_rank) then negate with np.errstate(invalid='ignore'): powered = np.power(np.abs(base_rank), power_rank) * np.sign(base_rank) alpha = -powered result[compute_mask] = alpha[compute_mask] result_valid = exists & valid & ~np.isnan(result) return TaggedArray( value=result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #71: Close-amount correlation decay rank vs price difference decay rank max. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_071.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Close-amount correlation decay ts_rank close_tsrank = ts_rank(close, self._close_rank_window) amount_mean = ts_mean(amount, self._amount_window) amount_tsrank = ts_rank(amount_mean, self._amount_rank_window) corr = ts_corr(close_tsrank, amount_tsrank, self._corr_window) corr_decayed = ts_decayed_linear(corr, self._decay_window1) first_part = ts_rank(corr_decayed, self._corr_final_window) # Part 2: Price difference squared decay ts_rank low_open_sum = low + open_ vwap_double = vwap + vwap price_diff = low_open_sum - vwap_double price_rank = cross_rank(price_diff) price_squared = price_rank**2 price_decayed = ts_decayed_linear(price_squared, self._decay_window2) second_part = ts_rank(price_decayed, self._price_final_window) # max(first_part, second_part) result = np.maximum(first_part, second_part) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #72: Mid price-amount correlation decay rank ratio. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_072.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1 (Numerator): Mid price-amount correlation decay rank mid_price = (high + low) / 2 amount_mean = ts_mean(amount, self._amount_window) corr1 = ts_corr(mid_price, amount_mean, self._corr_window1) decayed1 = ts_decayed_linear(corr1, self._decay_window1) numerator = cross_rank(decayed1) # Part 2 (Denominator): VWAP-volume ts_rank correlation decay rank vwap_tsrank = ts_rank(vwap, self._vwap_rank_window) volume_tsrank = ts_rank(volume, self._volume_rank_window) corr2 = ts_corr(vwap_tsrank, volume_tsrank, self._corr_window2) decayed2 = ts_decayed_linear(corr2, self._decay_window2) denominator = cross_rank(decayed2) # div(numerator, denominator) with np.errstate(divide="ignore", invalid="ignore"): result = np.where(denominator != 0, numerator / denominator, 0) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #73: VWAP delta decay rank vs weighted price change rate decay rank max. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_073.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) weight = self._weight # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: VWAP delta decay rank vwap_delta = ts_delta(vwap, self._vwap_delta_window) vwap_decayed = ts_decayed_linear(vwap_delta, self._decay_window1) first_part = cross_rank(vwap_decayed) # Part 2: Weighted open-low change rate decay ts_rank weighted_open_low = open_ * weight + low * (1 - weight) weighted_delta = ts_delta(weighted_open_low, self._price_delta_window) with np.errstate(divide="ignore", invalid="ignore"): change_rate = np.where( weighted_open_low != 0, weighted_delta / weighted_open_low, 0 ) neg_change_rate = change_rate * -1 rate_decayed = ts_decayed_linear(neg_change_rate, self._decay_window2) second_part = ts_rank(rate_decayed, self._rank_window) # max(first_part, second_part) max_result = np.maximum(first_part, second_part) # mul(max_result, -1) result = max_result * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #74: Close-amount correlation rank vs weighted high-VWAP volume correlation rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_074.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Close-amount correlation amount_mean = ts_mean(amount, self._amount_window) amount_sum = ts_sum(amount_mean, self._sum_window) close_corr = ts_corr(close, amount_sum, self._corr_window1) first_rank = cross_rank(close_corr) # Part 2: Weighted high-VWAP volume correlation weighted_high_vwap = high * weight + vwap * (1 - weight) weighted_rank = cross_rank(weighted_high_vwap) volume_rank = cross_rank(volume) weighted_corr = ts_corr(weighted_rank, volume_rank, self._corr_window2) second_rank = cross_rank(weighted_corr) # lt(first_rank, second_rank) * -1 condition = (first_rank < second_rank).astype(float) result = condition * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #75: VWAP-volume correlation rank vs low-amount rank correlation rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_075.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: VWAP-volume correlation vwap_corr = ts_corr(vwap, volume, self._corr_window1) first_rank = cross_rank(vwap_corr) # Part 2: Low-amount rank correlation low_rank = cross_rank(low) amount_mean = ts_mean(amount, self._amount_window) amount_rank = cross_rank(amount_mean) low_corr = ts_corr(low_rank, amount_rank, self._corr_window2) second_rank = cross_rank(low_corr) # lt(first_rank, second_rank) result = (first_rank < second_rank).astype(float) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #76: VWAP delta decay rank vs demeaned low-amount correlation decay rank max. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_076.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: VWAP delta decay rank vwap_delta = ts_delta(vwap, self._delta_window) vwap_decayed = ts_decayed_linear(vwap_delta, self._decay_window1) first_part = cross_rank(vwap_decayed) # Part 2: Demeaned low-amount correlation decay ts_rank low_demeaned = demean(low) amount_mean = ts_mean(amount, self._amount_window) low_corr = ts_corr(low_demeaned, amount_mean, self._corr_window) corr_ranked = ts_rank(low_corr, self._rank_window1) corr_decayed = ts_decayed_linear(corr_ranked, self._decay_window2) second_part = ts_rank(corr_decayed, self._rank_window2) # max(first_part, second_part) max_result = np.maximum(first_part, second_part) # mul(max_result, -1) result = max_result * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #77: Price difference decay rank vs mid-amount correlation decay rank min. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_077.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Price difference decay rank mid_price = (high + low) / 2 mid_high_sum = mid_price + high vwap_high_sum = vwap + high price_diff = mid_high_sum - vwap_high_sum price_decayed = ts_decayed_linear(price_diff, self._decay_window1) first_rank = cross_rank(price_decayed) # Part 2: Mid price-amount correlation decay rank amount_mean = ts_mean(amount, self._amount_window) corr = ts_corr(mid_price, amount_mean, self._corr_window) corr_decayed = ts_decayed_linear(corr, self._decay_window2) second_rank = cross_rank(corr_decayed) # min(first_rank, second_rank) result = np.minimum(first_rank, second_rank) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #78: Weighted low-VWAP amount correlation rank power by VWAP-volume rank correlation rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_078.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Weighted low-VWAP amount sum correlation weighted_low_vwap = low * weight + vwap * (1 - weight) weighted_sum = ts_sum(weighted_low_vwap, self._sum_window) amount_mean = ts_mean(amount, self._amount_window) amount_sum = ts_sum(amount_mean, self._sum_window) first_corr = ts_corr(weighted_sum, amount_sum, self._corr_window1) base_rank = cross_rank(first_corr) # Part 2: VWAP-volume rank correlation vwap_rank = cross_rank(vwap) volume_rank = cross_rank(volume) second_corr = ts_corr(vwap_rank, volume_rank, self._corr_window2) power_rank = cross_rank(second_corr) # pow(base_rank, power_rank) with np.errstate(invalid="ignore"): result = np.power(base_rank, power_rank) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #79: Demeaned weighted close-open delta rank vs VWAP-amount ts\_rank correlation rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_079.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: Demeaned weighted close-open delta weighted_close_open = close * weight + open_ * (1 - weight) weighted_demeaned = demean(weighted_close_open) weighted_delta = ts_delta(weighted_demeaned, self._delta_window) first_rank = cross_rank(weighted_delta) # Part 2: VWAP-amount ts_rank correlation vwap_tsrank = ts_rank(vwap, self._vwap_rank_window) amount_mean = ts_mean(amount, self._amount_window) amount_tsrank = ts_rank(amount_mean, self._amount_rank_window) corr = ts_corr(vwap_tsrank, amount_tsrank, self._corr_window) second_rank = cross_rank(corr) # lt(first_rank, second_rank) result = (first_rank < second_rank).astype(float) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #80: Demeaned weighted open-high delta sign rank power by high-amount correlation ts\_rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_080.""" if isinstance(data, list) and len(data) >= 4: _open_d, _high_d, _low_d, _close_d = data[:4] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: Demeaned weighted open-high delta sign weighted_open_high = open_ * weight + high * (1 - weight) weighted_demeaned = demean(weighted_open_high) weighted_delta = ts_delta(weighted_demeaned, self._delta_window) delta_sign = np.sign(weighted_delta) base_rank = cross_rank(delta_sign) # Part 2: High-amount correlation ts_rank amount_mean = ts_mean(amount, self._amount_window) corr = ts_corr(high, amount_mean, self._corr_window) power_rank = ts_rank(corr, self._rank_window) # pow(base_rank, power_rank) with np.errstate(invalid="ignore"): powered = np.power(base_rank, power_rank) # mul(powered, -1) result = powered * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #81: VWAP-amount correlation product log rank vs VWAP-volume rank correlation rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_081.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_product(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanprod(arr[t - window + 1 : t + 1], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: VWAP-amount correlation product log rank amount_mean = ts_mean(amount, self._amount_window) amount_sum = ts_sum(amount_mean, self._sum_window) vwap_corr = ts_corr(vwap, amount_sum, self._corr_window1) corr_rank = cross_rank(vwap_corr) corr_powered = np.power(corr_rank, 4) powered_rank = cross_rank(corr_powered) product_result = ts_product(powered_rank, self._product_window) with np.errstate(divide="ignore", invalid="ignore"): log_result = np.log(np.maximum(product_result, 1e-10)) first_rank = cross_rank(log_result) # Part 2: VWAP-volume rank correlation vwap_rank = cross_rank(vwap) volume_rank = cross_rank(volume) second_corr = ts_corr(vwap_rank, volume_rank, self._corr_window2) second_rank = cross_rank(second_corr) # lt(first_rank, second_rank) * -1 condition = (first_rank < second_rank).astype(float) result = condition * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #82: Open delta decay rank vs demeaned volume-open correlation decay rank min. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_082.""" if isinstance(data, list) and len(data) >= 2: _open_d, _volume_d = data[:2] data = {"open": _open_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = open_.shape # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: Open delta decay rank open_delta = ts_delta(open_, self._delta_window) open_decayed = ts_decayed_linear(open_delta, self._decay_window1) first_part = cross_rank(open_decayed) # Part 2: Demeaned volume-open correlation decay ts_rank volume_demeaned = demean(volume) # Weighted open: open * 0.634196 + open * (1-0.634196) = open weighted_open = open_ corr = ts_corr(volume_demeaned, weighted_open, self._corr_window) corr_decayed = ts_decayed_linear(corr, self._decay_window2) second_part = ts_rank(corr_decayed, self._rank_window) # min(first_part, second_part) min_result = np.minimum(first_part, second_part) # mul(min_result, -1) result = min_result * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #83: Delayed range ratio rank times double volume rank ratio. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_083.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Helper functions def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def delay(arr, period): result = np.full_like(arr, np.nan) result[period:] = arr[:-period] return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Common calculation: range ratio high_low_range = high - low close_sum = ts_sum(close, self._mean_window) close_mean = close_sum / self._mean_window with np.errstate(divide="ignore", invalid="ignore"): range_ratio = np.where(close_mean != 0, high_low_range / close_mean, 0) # Numerator delayed_ratio = delay(range_ratio, self._delay_period) delayed_rank = cross_rank(delayed_ratio) volume_rank = cross_rank(volume) double_volume_rank = cross_rank(volume_rank) numerator = delayed_rank * double_volume_rank # Denominator vwap_close_diff = vwap - close with np.errstate(divide="ignore", invalid="ignore"): denominator = np.where(vwap_close_diff != 0, range_ratio / vwap_close_diff, 0) # numerator / denominator with np.errstate(divide="ignore", invalid="ignore"): result = np.where(denominator != 0, numerator / denominator, 0) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #84: VWAP max difference ts\_rank power by close delta. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_084.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Helper functions def ts_max(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] if not np.all(np.isnan(col)): result[t, s] = np.nanmax(col) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result # Base: VWAP - max VWAP ts_rank vwap_max = ts_max(vwap, self._max_window) vwap_diff = vwap - vwap_max base = ts_rank(vwap_diff, self._rank_window) # Exponent: close delta power = ts_delta(close, self._delta_window) # pow(base, power) with np.errstate(invalid="ignore"): result = np.power(base, power) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #85: Weighted high-close amount correlation rank power by mid-volume ts\_rank correlation rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_085.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Base: Weighted high-close amount correlation rank weighted_high_close = high * weight + close * (1 - weight) amount_mean = ts_mean(amount, self._amount_window) first_corr = ts_corr(weighted_high_close, amount_mean, self._corr_window1) base = cross_rank(first_corr) # Exponent: Mid price-volume ts_rank correlation rank mid_price = (high + low) / 2 mid_tsrank = ts_rank(mid_price, self._mid_rank_window) vol_tsrank = ts_rank(volume, self._vol_rank_window) second_corr = ts_corr(mid_tsrank, vol_tsrank, self._corr_window2) power = cross_rank(second_corr) # pow(base, power) with np.errstate(invalid="ignore"): result = np.power(base, power) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #86: Close-amount correlation ts\_rank vs price sum difference rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_086.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Close-amount correlation ts_rank amount_mean = ts_mean(amount, self._amount_window) amount_sum = ts_sum(amount_mean, self._sum_window) close_corr = ts_corr(close, amount_sum, self._corr_window) first_part = ts_rank(close_corr, self._rank_window) # Part 2: Price sum difference rank open_close_sum = open_ + close vwap_open_sum = vwap + open_ price_diff = open_close_sum - vwap_open_sum second_part = cross_rank(price_diff) # lt(first_part, second_part) * -1 condition = (first_part < second_part).astype(float) result = condition * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #87: Weighted close-VWAP delta decay rank vs demeaned amount-close correlation abs decay rank max. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_087.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: Weighted close-VWAP delta decay rank weighted_close_vwap = close * weight + vwap * (1 - weight) weighted_delta = ts_delta(weighted_close_vwap, self._delta_window) delta_decayed = ts_decayed_linear(weighted_delta, self._decay_window1) first_part = cross_rank(delta_decayed) # Part 2: Demeaned amount-close correlation abs decay ts_rank amount_mean = ts_mean(amount, self._amount_window) amount_demeaned = demean(amount_mean) amount_corr = ts_corr(amount_demeaned, close, self._corr_window) abs_corr = np.abs(amount_corr) corr_decayed = ts_decayed_linear(abs_corr, self._decay_window2) second_part = ts_rank(corr_decayed, self._rank_window) # max(first_part, second_part) max_result = np.maximum(first_part, second_part) # mul(max_result, -1) result = max_result * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #88: Price rank sum difference decay rank vs close-amount ts\_rank correlation decay rank min. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_088.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Price rank sum difference decay rank open_rank = cross_rank(open_) low_rank = cross_rank(low) high_rank = cross_rank(high) close_rank = cross_rank(close) open_low_sum = open_rank + low_rank high_close_sum = high_rank + close_rank rank_diff = open_low_sum - high_close_sum diff_decayed = ts_decayed_linear(rank_diff, self._decay_window1) first_part = cross_rank(diff_decayed) # Part 2: Close-amount ts_rank correlation decay ts_rank close_tsrank = ts_rank(close, self._close_rank_window) amount_mean = ts_mean(amount, self._amount_window) amount_tsrank = ts_rank(amount_mean, self._amount_rank_window) corr = ts_corr(close_tsrank, amount_tsrank, self._corr_window) corr_decayed = ts_decayed_linear(corr, self._decay_window2) second_part = ts_rank(corr_decayed, self._final_rank_window) # min(first_part, second_part) result = np.minimum(first_part, second_part) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #89: Weighted low-amount correlation decay rank minus demeaned VWAP delta decay rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_089.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: Low-amount correlation decay ts_rank # Weighted low: low * 0.967285 + low * (1-0.967285) = low weighted_low = low amount_mean = ts_mean(amount, self._amount_window) low_corr = ts_corr(weighted_low, amount_mean, self._corr_window) low_decayed = ts_decayed_linear(low_corr, self._decay_window1) first_part = ts_rank(low_decayed, self._rank_window1) # Part 2: Demeaned VWAP delta decay ts_rank vwap_demeaned = demean(vwap) vwap_delta = ts_delta(vwap_demeaned, self._delta_window) vwap_decayed = ts_decayed_linear(vwap_delta, self._decay_window2) second_part = ts_rank(vwap_decayed, self._rank_window2) # sub(first_part, second_part) result = first_part - second_part # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #90: Close max difference rank power by demeaned amount-low correlation ts\_rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_090.""" if isinstance(data, list) and len(data) >= 3: _low_d, _close_d, _volume_d = data[:3] data = {"low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close # Helper functions def ts_max(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] if not np.all(np.isnan(col)): result[t, s] = np.nanmax(col) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Base: Close - max close rank close_max = ts_max(close, self._max_window) close_diff = close - close_max base = cross_rank(close_diff) # Exponent: Demeaned amount-low correlation ts_rank amount_mean = ts_mean(amount, self._amount_window) amount_demeaned = demean(amount_mean) amount_low_corr = ts_corr(amount_demeaned, low, self._corr_window) power = ts_rank(amount_low_corr, self._rank_window) # pow(base, power) with np.errstate(invalid="ignore"): powered = np.power(base, power) # mul(powered, -1) result = powered * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #91: Double decayed close-volume correlation ts\_rank minus VWAP-amount correlation decay rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_091.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: Double decayed demeaned close-volume correlation ts_rank close_demeaned = demean(close) close_vol_corr = ts_corr(close_demeaned, volume, self._corr_window1) first_decayed = ts_decayed_linear(close_vol_corr, self._decay_window1) second_decayed = ts_decayed_linear(first_decayed, self._decay_window2) first_part = ts_rank(second_decayed, self._rank_window1) # Part 2: VWAP-amount correlation decay rank amount_mean = ts_mean(amount, self._amount_window) vwap_amount_corr = ts_corr(vwap, amount_mean, self._corr_window2) vwap_decayed = ts_decayed_linear(vwap_amount_corr, self._decay_window3) second_part = cross_rank(vwap_decayed) # sub(first_part, second_part) * -1 diff = first_part - second_part result = diff * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #92: Mid-close vs low-open comparison decay rank min with low-amount rank correlation decay rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_092.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Price comparison condition decay ts_rank mid_price = (high + low) / 2 mid_close_sum = mid_price + close low_open_sum = low + open_ price_condition = (mid_close_sum < low_open_sum).astype(float) condition_decayed = ts_decayed_linear(price_condition, self._decay_window1) first_part = ts_rank(condition_decayed, self._rank_window1) # Part 2: Low-amount rank correlation decay ts_rank low_rank = cross_rank(low) amount_mean = ts_mean(amount, self._amount_window) amount_rank = cross_rank(amount_mean) low_amount_corr = ts_corr(low_rank, amount_rank, self._corr_window) corr_decayed = ts_decayed_linear(low_amount_corr, self._decay_window2) second_part = ts_rank(corr_decayed, self._rank_window2) # min(first_part, second_part) result = np.minimum(first_part, second_part) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #93: Demeaned VWAP-amount correlation decay ts\_rank divided by weighted close-VWAP delta decay rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_093.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Numerator: Demeaned VWAP-amount correlation decay ts_rank vwap_demeaned = demean(vwap) amount_mean = ts_mean(amount, self._amount_window) vwap_amount_corr = ts_corr(vwap_demeaned, amount_mean, self._corr_window) corr_decayed = ts_decayed_linear(vwap_amount_corr, self._decay_window1) numerator = ts_rank(corr_decayed, self._rank_window) # Denominator: Weighted close-VWAP delta decay rank weighted_close_vwap = close * weight + vwap * (1 - weight) weighted_delta = ts_delta(weighted_close_vwap, self._delta_window) delta_decayed = ts_decayed_linear(weighted_delta, self._decay_window2) denominator = cross_rank(delta_decayed) # div(numerator, denominator) with np.errstate(divide="ignore", invalid="ignore"): result = np.where(denominator != 0, numerator / denominator, 0) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #94: VWAP-min VWAP difference rank power by VWAP-amount ts\_rank correlation ts\_rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_094.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_min(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] if not np.all(np.isnan(col)): result[t, s] = np.nanmin(col) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Base: VWAP - min VWAP difference rank vwap_min = ts_min(vwap, self._min_window) vwap_diff = vwap - vwap_min base = cross_rank(vwap_diff) # Exponent: VWAP-amount ts_rank correlation ts_rank vwap_tsrank = ts_rank(vwap, self._vwap_rank_window) amount_mean = ts_mean(amount, self._amount_window) amount_tsrank = ts_rank(amount_mean, self._amount_rank_window) corr_result = ts_corr(vwap_tsrank, amount_tsrank, self._corr_window) power = ts_rank(corr_result, self._final_rank_window) # pow(base, power) * -1 with np.errstate(invalid="ignore"): powered = np.power(base, power) result = powered * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #95: Open-min open difference rank less than mid-amount correlation rank power ts\_rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_095.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close # Helper functions def ts_min(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] if not np.all(np.isnan(col)): result[t, s] = np.nanmin(col) return result def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Open - min open difference rank open_min = ts_min(open_, self._min_window) open_diff = open_ - open_min first_part = cross_rank(open_diff) # Part 2: Mid-amount correlation rank power ts_rank mid_price = (high + low) / 2 mid_sum = ts_sum(mid_price, self._sum_window) amount_mean = ts_mean(amount, self._amount_window) amount_sum = ts_sum(amount_mean, self._sum_window) corr_result = ts_corr(mid_sum, amount_sum, self._corr_window) corr_rank = cross_rank(corr_result) with np.errstate(over="ignore", invalid="ignore"): corr_powered = np.power(corr_rank, self._power_exp) corr_powered = np.nan_to_num(corr_powered, nan=0.0, posinf=0.0, neginf=0.0) second_part = ts_rank(corr_powered, self._final_rank_window) # lt(first_part, second_part) result = (first_part < second_part).astype(float) # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #96: VWAP-volume rank correlation decay ts\_rank vs close-amount ts\_rank correlation argmax decay ts\_rank max. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_096.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_argmax(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] if np.all(np.isnan(col)): result[t, s] = np.nan else: result[t, s] = np.nanargmax(col) return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: VWAP-volume rank correlation decay ts_rank vwap_rank = cross_rank(vwap) volume_rank = cross_rank(volume) vwap_vol_corr = ts_corr(vwap_rank, volume_rank, self._corr_window1) first_decayed = ts_decayed_linear(vwap_vol_corr, self._decay_window1) first_part = ts_rank(first_decayed, self._rank_window1) # Part 2: Close-amount ts_rank correlation argmax decay ts_rank close_tsrank = ts_rank(close, self._close_rank_window) amount_mean = ts_mean(amount, self._amount_window) amount_tsrank = ts_rank(amount_mean, self._amount_rank_window) close_amount_corr = ts_corr(close_tsrank, amount_tsrank, self._corr_window2) corr_argmax = ts_argmax(close_amount_corr, self._argmax_window) second_decayed = ts_decayed_linear(corr_argmax, self._decay_window2) second_part = ts_rank(second_decayed, self._rank_window2) # max(first_part, second_part) * -1 max_result = np.maximum(first_part, second_part) result = max_result * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #97: Demeaned weighted low-VWAP delta decay rank minus low-amount ts\_rank correlation decay ts\_rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_097.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close weight = self._weight # Helper functions def ts_delta(arr, window): result = np.full_like(arr, np.nan) result[window:] = arr[window:] - arr[:-window] return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result # Part 1: Demeaned weighted low-VWAP delta decay rank weighted_low_vwap = low * weight + vwap * (1 - weight) weighted_demeaned = demean(weighted_low_vwap) weighted_delta = ts_delta(weighted_demeaned, self._delta_window) delta_decayed = ts_decayed_linear(weighted_delta, self._decay_window1) first_part = cross_rank(delta_decayed) # Part 2: Low-amount ts_rank correlation decay ts_rank low_tsrank = ts_rank(low, self._low_rank_window) amount_mean = ts_mean(amount, self._amount_window) amount_tsrank = ts_rank(amount_mean, self._amount_rank_window) low_amount_corr = ts_corr(low_tsrank, amount_tsrank, self._corr_window) corr_ranked = ts_rank(low_amount_corr, self._corr_rank_window) corr_decayed = ts_decayed_linear(corr_ranked, self._decay_window2) second_part = ts_rank(corr_decayed, self._final_rank_window) # sub(first_part, second_part) * -1 diff = first_part - second_part result = diff * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #98: VWAP-amount correlation decay rank minus open-amount rank correlation argmin decay rank. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_098.""" if isinstance(data, list) and len(data) >= 5: _open_d, _high_d, _low_d, _close_d, _volume_d = data[:5] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate VWAP typical_price = (high + low + close) / 3 with np.errstate(divide='ignore', invalid='ignore'): cum_vol = np.cumsum(volume, axis=0) vwap = np.where(cum_vol > 0, np.cumsum(typical_price * volume, axis=0) / cum_vol, 0.0) # Calculate amount amount = volume * close # Helper functions def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_rank(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] valid = ~np.isnan(col) if np.sum(valid) > 0: result[t, s] = np.sum(col[valid] <= col[-1]) / np.sum(valid) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_argmin(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] if np.all(np.isnan(col)): result[t, s] = np.nan else: result[t, s] = np.nanargmin(col) return result def ts_decayed_linear(arr, window): weights = np.arange(1, window + 1, dtype=float) weights = weights / weights.sum() result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] result[t] = np.nansum(window_data * weights[:, np.newaxis], axis=0) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: VWAP-amount correlation decay rank amount_mean_5 = ts_mean(amount, self._amount_window1) amount_sum = ts_sum(amount_mean_5, self._sum_window) vwap_corr = ts_corr(vwap, amount_sum, self._corr_window1) vwap_decayed = ts_decayed_linear(vwap_corr, self._decay_window1) first_part = cross_rank(vwap_decayed) # Part 2: Open-amount rank correlation argmin decay rank open_rank = cross_rank(open_) amount_mean_15 = ts_mean(amount, self._amount_window2) amount_rank = cross_rank(amount_mean_15) open_amount_corr = ts_corr(open_rank, amount_rank, self._corr_window2) corr_argmin = ts_argmin(open_amount_corr, self._argmin_window) argmin_ranked = ts_rank(corr_argmin, self._argmin_rank_window) argmin_decayed = ts_decayed_linear(argmin_ranked, self._decay_window2) second_part = cross_rank(argmin_decayed) # sub(first_part, second_part) result = first_part - second_part # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #99: Mid-amount sum correlation rank less than low-volume correlation rank comparison. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_099.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close # Helper functions def ts_sum(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nansum(arr[t - window + 1 : t + 1], axis=0) return result def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result # Part 1: Mid-amount sum correlation rank mid_price = (high + low) / 2 mid_sum = ts_sum(mid_price, self._sum_window) amount_mean = ts_mean(amount, self._amount_window) amount_sum = ts_sum(amount_mean, self._sum_window) mid_amount_corr = ts_corr(mid_sum, amount_sum, self._corr_window1) first_rank = cross_rank(mid_amount_corr) # Part 2: Low-volume correlation rank low_vol_corr = ts_corr(low, volume, self._corr_window2) second_rank = cross_rank(low_vol_corr) # lt(first_rank, second_rank) * -1 condition = (first_rank < second_rank).astype(float) result = condition * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #100: Complex multi-demeaned price position-volume and correlation factor. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_100.""" if isinstance(data, list) and len(data) >= 4: _high_d, _low_d, _close_d, _volume_d = data[:4] data = {"high": _high_d.value, "low": _low_d.value, "close": _close_d.value, "volume": _volume_d.value} high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) volume = np.array(data["volume"]) n_timepoints, n_symbols = close.shape # Calculate amount amount = volume * close # Helper functions def ts_mean(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): result[t] = np.nanmean(arr[t - window + 1 : t + 1], axis=0) return result def ts_corr(arr1, arr2, window): result = np.full_like(arr1, np.nan) for t in range(window - 1, n_timepoints): for s in range(n_symbols): x = arr1[t - window + 1 : t + 1, s] y = arr2[t - window + 1 : t + 1, s] valid = ~(np.isnan(x) | np.isnan(y)) if np.sum(valid) > 2: corr = np.corrcoef(x[valid], y[valid])[0, 1] result[t, s] = corr if not np.isnan(corr) else 0 return result def ts_argmin(arr, window): result = np.full_like(arr, np.nan) for t in range(window - 1, n_timepoints): window_data = arr[t - window + 1 : t + 1] for s in range(n_symbols): col = window_data[:, s] if np.all(np.isnan(col)): result[t, s] = np.nan else: result[t, s] = np.nanargmin(col) return result def cross_rank(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: ranked = (np.argsort(np.argsort(row)) + 1) / len(row) result[t] = ranked return result def demean(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: result[t] = row - np.nanmean(row) return result def scale(arr): result = np.full_like(arr, np.nan) for t in range(n_timepoints): row = arr[t] valid = ~np.isnan(row) if np.sum(valid) > 0: abs_sum = np.nansum(np.abs(row)) if abs_sum > 0: result[t] = row / abs_sum else: result[t] = row return result # Part 1: Multi-demeaned price position-volume # Williams %R like calculation close_low = close - low high_close = high - close high_low = high - low with np.errstate(divide="ignore", invalid="ignore"): williams_r_num = close_low - high_close williams_r = np.where(high_low != 0, williams_r_num / high_low, 0) # mul(williams_r, volume) wr_volume = williams_r * volume # rank(wr_volume) wr_rank = cross_rank(wr_volume) # Double demeaning first_demean = demean(wr_rank) second_demean = demean(first_demean) # scale and mul(1.5, ...) first_scaled = scale(second_demean) first_part = self._scale_factor * first_scaled # Part 2: Correlation and argmin difference amount_mean = ts_mean(amount, self._amount_window) amount_rank = cross_rank(amount_mean) close_amount_corr = ts_corr(close, amount_rank, self._corr_window) close_argmin = ts_argmin(close, self._argmin_window) argmin_rank = cross_rank(close_argmin) corr_diff = close_amount_corr - argmin_rank corr_demeaned = demean(corr_diff) second_part = scale(corr_demeaned) # Part 3: Volume ratio with np.errstate(divide="ignore", invalid="ignore"): volume_ratio = np.where(amount_mean != 0, volume / amount_mean, 0) # Final calculation main_diff = first_part - second_part product = main_diff * volume_ratio result = product * -1 # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` Alpha #101: Price change divided by price range. ```python theme={null} def compute(self, data, timestamp=None, context=None) -> TaggedArray: """Calculate Alpha 101_101.""" if isinstance(data, list) and len(data) >= 4: _open_d, _high_d, _low_d, _close_d = data[:4] data = {"open": _open_d.value, "high": _high_d.value, "low": _low_d.value, "close": _close_d.value} open_ = np.array(data["open"]) high = np.array(data["high"]) low = np.array(data["low"]) close = np.array(data["close"]) n_timepoints, n_symbols = close.shape # sub(close, open) - price change price_change = close - open_ # sub(high, low) - price range price_range = high - low # add(price_range, epsilon) - prevent division by zero adjusted_range = price_range + self._epsilon # div(price_change, adjusted_range) - normalized return result = price_change / adjusted_range # Fill NaN with 0 result = np.nan_to_num(result, nan=0.0) final_result = result[-1] exists = np.array([True] * n_symbols) result_valid = ~np.isnan(final_result) return TaggedArray( value=final_result, exists=exists, valid=result_valid, updated=np.ones(n_symbols, dtype=bool), ) ``` ## Related Pages Hand-crafted alpha signals Cross-sectional risk factors