From 7b01c5cae7830d2b75c5566443f5a4559b5b2f40 Mon Sep 17 00:00:00 2001 From: Charles Young Date: Tue, 9 Feb 2021 20:30:26 +0800 Subject: [PATCH] Add an implementation of Enhanced Indexing to optimizer.py --- qlib/.DS_Store | Bin 0 -> 6148 bytes qlib/portfolio/optimizer.py | 129 ++++++++++++++++++++++++++++++++---- 2 files changed, 117 insertions(+), 12 deletions(-) create mode 100644 qlib/.DS_Store diff --git a/qlib/.DS_Store b/qlib/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..3b196d96a164ebf17b658d6f6d8c5ef19fb26c8c GIT binary patch literal 6148 zcmeHK%Wl&^6upx=txW}FQHgGlykS>WX{it_kU}Vn?vR3D0VuU=(^|NmD0Z;g5R^S% z0N;SbckloAHxWIzRhp0nDE)t`24cQXodQJ=0 zG9@>FLbZ`lK|@^IG+JQW0;_;k;9paK=kB_ivk7HXoxf)XsXPi(8G^yf!y}~apUU77 zd2~b)F>erctY&;O;%g&f%|sM5Mod_5U)*l69f~NxdJ;5v9~--ZHYJy`PvXJrW|?Rkd0?VRkk)wz)B!- zgQ;dimFZHcy5udogvWbXZ&~oHsPb`pshPpyu3yTu1gn5m;D1wq*9RYov1f3uQ7s+F z)DZyapja7v{#oD{-(b(+Tq9~=LWcr%s4zzip~F$`8(z=gT%!&rVGbX{JXx3%icn8S z`@W)+=xMaMRlq7xS71foc6k4Pa`E}U9%S#V0#<=5rGRku2m5_|lG$4qK92WVAL$T@ qjd^p8Dg>F?j+Mh(@g|Zo)cNcHdj{tkQ3A6+0!jv(Sq1*80>1zhm={X` literal 0 HcmV?d00001 diff --git a/qlib/portfolio/optimizer.py b/qlib/portfolio/optimizer.py index 0e7d27254..e04923ed6 100644 --- a/qlib/portfolio/optimizer.py +++ b/qlib/portfolio/optimizer.py @@ -28,13 +28,13 @@ class PortfolioOptimizer: OPT_INV = "inv" def __init__( - self, - method: str = "inv", - lamb: float = 0, - delta: float = 0, - alpha: float = 0.0, - scale_alpha: bool = True, - tol: float = 1e-8, + self, + method: str = "inv", + lamb: float = 0, + delta: float = 0, + alpha: float = 0.0, + scale_alpha: bool = True, + tol: float = 1e-8, ): """ Args: @@ -59,10 +59,10 @@ class PortfolioOptimizer: self.tol = tol def __call__( - self, - S: Union[np.ndarray, pd.DataFrame], - u: Optional[Union[np.ndarray, pd.Series]] = None, - w0: Optional[Union[np.ndarray, pd.Series]] = None, + self, + S: Union[np.ndarray, pd.DataFrame], + u: Optional[Union[np.ndarray, pd.Series]] = None, + w0: Optional[Union[np.ndarray, pd.Series]] = None, ) -> Union[np.ndarray, pd.Series]: """ Args: @@ -151,7 +151,7 @@ class PortfolioOptimizer: return self._solve(len(S), self._get_objective_gmv(S), *self._get_constrains(w0)) def _optimize_mvo( - self, S: np.ndarray, u: Optional[np.ndarray] = None, w0: Optional[np.ndarray] = None + self, S: np.ndarray, u: Optional[np.ndarray] = None, w0: Optional[np.ndarray] = None ) -> np.ndarray: """optimize mean-variance portfolio @@ -256,3 +256,108 @@ class PortfolioOptimizer: warnings.warn(f"optimization not success ({sol.status})") return sol.x + + +class EnhancedIndexingOptimizer: + """ + Portfolio Optimizer with Enhanced Indexing + + Note: + This optimizer always assumes full investment and no-shorting. + """ + + START_FROM_W0 = 'w0' + START_FROM_BENCH = 'benchmark' + DO_NOT_START_FROM = '' + + def __init__(self, lamb: float = 10, delta: float = 0.4, bench_dev: float = 0.01, inds_dev: float = 0.01, + scale_alpha=True, verbose: bool = False, warm_start: str = '', max_iters: int = 10000): + """ + Args: + lamb (float): risk aversion parameter (larger `lamb` means less focus on return) + delta (float): turnover rate limit + bench_dev (float): benchmark deviation limit + inds_dev (float): industry deviation limit + verbose (bool): if print detailed information about the solver + warm_start (str): whether try to warm start (`w0`/`benchmark`/``) + (https://www.cvxpy.org/tutorial/advanced/index.html#warm-start) + """ + + assert lamb >= 0, "risk aversion parameter `lamb` should be positive" + self.lamb = lamb + + assert delta >= 0, "turnover limit `delta` should be positive" + self.delta = delta + + assert bench_dev >= 0, "benchmark deviation limit `bench_dev` should be positive" + self.bench_dev = bench_dev + + assert inds_dev >= 0, "industry deviation limit `inds_dev` should be positive" + self.inds_dev = inds_dev + + assert warm_start in [self.DO_NOT_START_FROM, self.START_FROM_W0, + self.START_FROM_BENCH], "illegal warm start option" + self.start_from_w0 = (warm_start == self.START_FROM_W0) + self.start_from_bench = (warm_start == self.START_FROM_BENCH) + + self.scale_alpha = scale_alpha + self.verbose = verbose + self.max_iters = max_iters + + def __call__(self, u: np.ndarray, F: np.ndarray, covB: np.ndarray, varU: np.ndarray, w0: np.ndarray, + w_bench: np.ndarray, inds_onehot: np.ndarray + ) -> Union[np.ndarray, pd.Series]: + """ + Args: + u (np.ndarray): expected returns (a.k.a., alpha) + F, covB, varU (np.ndarray): see StructuredCovEstimator + w0 (np.ndarray): initial weights (for turnover control) + w_bench (np.ndarray): benchmark weights + inds_onehot (np.ndarray): industry (onehot) + + Returns: + np.ndarray or pd.Series: optimized portfolio allocation + """ + # scale alpha to match volatility + if self.scale_alpha: + u = u / u.std() + x_variance = np.mean(np.diag(F @ covB @ F.T) + varU) + u *= x_variance ** 0.5 + + w = cp.Variable(len(u)) # num_assets + v = w @ F # num_factors + ret = w @ u + risk = cp.quad_form(v, covB) + cp.sum(cp.multiply(varU, w ** 2)) + obj = cp.Maximize(ret - self.lamb * risk) + d_bench = w - w_bench + d_inds = d_bench @ inds_onehot + cons = [ + w >= 0, + cp.sum(w) == 1, + d_bench >= -self.bench_dev, + d_bench <= self.bench_dev, + d_inds >= -self.inds_dev, + d_inds <= self.inds_dev + ] + if w0 is not None: + turnover = cp.sum(cp.abs(w - w0)) + cons.append(turnover <= self.delta) + + warm_start = False + if self.start_from_w0: + if w0 is None: + print('Warning: try warm start with w0, but w0 is `None`.') + else: + w.value = w0 + warm_start = True + elif self.start_from_bench: + w.value = w_bench + warm_start = True + + prob = cp.Problem(obj, cons) + prob.solve(solver=cp.SCS, verbose=self.verbose, warm_start=warm_start, max_iters=self.max_iters) + + if prob.status != 'optimal': + print('Warning: solve failed.', prob.status) + + return np.asarray(w.value)