|
1 | | -import numpy as np |
2 | | -import torch |
3 | | -import torch.nn as nn |
4 | | -from torch.utils.data import DataLoader |
5 | | -import jsonargparse |
6 | | -from asset_pricing_benchmark import mu_f_array |
7 | | -from typing import List, Optional |
8 | | - |
9 | | - |
10 | | -def asset_pricing_neural( |
11 | | - r: float = 0.1, |
12 | | - c: float = 0.02, |
13 | | - g: float = -0.2, |
14 | | - x_0: float = 0.01, |
15 | | - train_T: float = 40.0, |
16 | | - train_points: int = 41, |
17 | | - test_T: float = 50.0, |
18 | | - test_points: int = 100, |
19 | | - train_points_list: Optional[List[float]] = None, |
20 | | - seed=123, |
21 | | -): |
22 | | - # if passing in `train_points` then doesn't us a grid. Otherwise, uses linspace |
23 | | - if train_points_list is None: |
24 | | - train_data = torch.tensor( |
25 | | - np.linspace(0, train_T, train_points), dtype=torch.float32 |
26 | | - ) |
27 | | - else: |
28 | | - train_data = torch.tensor(np.array(train_points_list), dtype=torch.float32) |
29 | | - train_data = train_data.unsqueeze(dim=1) |
30 | | - test_data = torch.tensor(np.linspace(0, test_T, test_points), dtype=torch.float32) |
31 | | - test_data = test_data.unsqueeze(dim=1) |
32 | | - |
33 | | - train = DataLoader(train_data, batch_size=len(train_data), shuffle=False) |
34 | | - |
35 | | - def derivative_back(model, t): # backward differencing |
36 | | - epsilon = 1.0e-8 |
37 | | - sqrt_eps = np.sqrt(epsilon) |
38 | | - return (model(t) - model(t - sqrt_eps)) / sqrt_eps |
39 | | - |
40 | | - # Dividends |
41 | | - def x(i): |
42 | | - return (x_0 + (c / g)) * np.exp(g * i) - (c / g) |
43 | | - |
44 | | - def G(model, t): |
45 | | - mu = model(t) |
46 | | - dmudt = r * mu - x(t) |
47 | | - return dmudt |
48 | | - |
49 | | - torch.manual_seed(seed) |
50 | | - |
51 | | - class NN(nn.Module): |
52 | | - def __init__( |
53 | | - self, |
54 | | - dim_hidden=128, |
55 | | - ): |
56 | | - super().__init__() |
57 | | - self.dim_hidden = dim_hidden |
58 | | - self.q = nn.Sequential( |
59 | | - nn.Linear(1, dim_hidden, bias=True), |
60 | | - nn.Tanh(), |
61 | | - nn.Linear(dim_hidden, dim_hidden, bias=True), |
62 | | - nn.Tanh(), |
63 | | - nn.Linear(dim_hidden, dim_hidden, bias=True), |
64 | | - nn.Tanh(), |
65 | | - nn.Linear(dim_hidden, dim_hidden, bias=True), |
66 | | - nn.Tanh(), |
67 | | - nn.Linear(dim_hidden, 1), |
68 | | - nn.Softplus(beta=1.0), # To make sure price stays positive |
69 | | - ) |
70 | | - |
71 | | - def forward(self, x): |
72 | | - return self.q(x) |
73 | | - |
74 | | - q_hat = NN() |
75 | | - learning_rate = 1e-3 |
76 | | - optimizer = torch.optim.Adam(q_hat.parameters(), lr=learning_rate) |
77 | | - scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=100, gamma=0.8) |
78 | | - num_epochs = 1000 |
79 | | - |
80 | | - for epoch in range(num_epochs): |
81 | | - for i, time in enumerate(train): |
82 | | - |
83 | | - res_ode = derivative_back(q_hat, time) - G(q_hat, time) |
84 | | - res_p_dot = res_ode[:, 0] |
85 | | - |
86 | | - loss = res_p_dot.pow(2).mean() |
87 | | - |
88 | | - optimizer.zero_grad() |
89 | | - loss.backward() |
90 | | - |
91 | | - optimizer.step() |
92 | | - scheduler.step() |
93 | | - |
94 | | - # Generate test_data and compare to the benchmark |
95 | | - mu_benchmark = mu_f_array(np.array(test_data), c, g, r, x_0) |
96 | | - mu_test = np.array(q_hat(test_data)[:, [0]].detach()) |
97 | | - |
98 | | - mu_rel_error = np.abs(mu_benchmark - mu_test) / mu_benchmark |
99 | | - print(f"E(|rel_error(p)|) = {mu_rel_error.mean()}") |
100 | | - return { |
101 | | - "t_train": train_data, |
102 | | - "t_test": test_data, |
103 | | - "p_test": mu_test, |
104 | | - "p_benchmark": mu_benchmark, |
105 | | - "p_rel_error": mu_rel_error, |
106 | | - "neural_net_solution": q_hat, # interpolator |
107 | | - } |
108 | | - |
109 | | - |
110 | | -if __name__ == "__main__": |
111 | | - jsonargparse.CLI(asset_pricing_neural) |
| 1 | +import numpy as np |
| 2 | +import torch |
| 3 | +import torch.nn as nn |
| 4 | +from torch.utils.data import DataLoader |
| 5 | +import jsonargparse |
| 6 | +from asset_pricing_benchmark import mu_f_array |
| 7 | +from typing import List, Optional |
| 8 | + |
| 9 | + |
| 10 | +def asset_pricing_neural( |
| 11 | + r: float = 0.1, |
| 12 | + c: float = 0.02, |
| 13 | + g: float = -0.2, |
| 14 | + x_0: float = 0.01, |
| 15 | + train_T: float = 40.0, |
| 16 | + train_points: int = 41, |
| 17 | + test_T: float = 50.0, |
| 18 | + test_points: int = 100, |
| 19 | + train_points_list: Optional[List[float]] = None, |
| 20 | + seed=123, |
| 21 | +): |
| 22 | + # if passing in `train_points` then doesn't us a grid. Otherwise, uses linspace |
| 23 | + if train_points_list is None: |
| 24 | + train_data = torch.tensor( |
| 25 | + np.linspace(0, train_T, train_points), dtype=torch.float32 |
| 26 | + ) |
| 27 | + else: |
| 28 | + train_data = torch.tensor(np.array(train_points_list), dtype=torch.float32) |
| 29 | + train_data = train_data.unsqueeze(dim=1) |
| 30 | + test_data = torch.tensor(np.linspace(0, test_T, test_points), dtype=torch.float32) |
| 31 | + test_data = test_data.unsqueeze(dim=1) |
| 32 | + |
| 33 | + train = DataLoader(train_data, batch_size=len(train_data), shuffle=False) |
| 34 | + |
| 35 | + def derivative_back(model, t): # backward differencing |
| 36 | + epsilon = 1.0e-8 |
| 37 | + sqrt_eps = np.sqrt(epsilon) |
| 38 | + return (model(t) - model(t - sqrt_eps)) / sqrt_eps |
| 39 | + |
| 40 | + # Dividends |
| 41 | + def x(i): |
| 42 | + return (x_0 + (c / g)) * np.exp(g * i) - (c / g) |
| 43 | + |
| 44 | + def G(model, t): |
| 45 | + mu = model(t) |
| 46 | + dmudt = r * mu - x(t) |
| 47 | + return dmudt |
| 48 | + |
| 49 | + torch.manual_seed(seed) |
| 50 | + |
| 51 | + class NN(nn.Module): |
| 52 | + def __init__( |
| 53 | + self, |
| 54 | + dim_hidden=128, |
| 55 | + ): |
| 56 | + super().__init__() |
| 57 | + self.dim_hidden = dim_hidden |
| 58 | + self.q = nn.Sequential( |
| 59 | + nn.Linear(1, dim_hidden, bias=True), |
| 60 | + nn.Tanh(), |
| 61 | + nn.Linear(dim_hidden, dim_hidden, bias=True), |
| 62 | + nn.Tanh(), |
| 63 | + nn.Linear(dim_hidden, dim_hidden, bias=True), |
| 64 | + nn.Tanh(), |
| 65 | + nn.Linear(dim_hidden, dim_hidden, bias=True), |
| 66 | + nn.Tanh(), |
| 67 | + nn.Linear(dim_hidden, 1), |
| 68 | + nn.Softplus(beta=1.0), # To make sure price stays positive |
| 69 | + ) |
| 70 | + |
| 71 | + def forward(self, x): |
| 72 | + return self.q(x) |
| 73 | + |
| 74 | + q_hat = NN() |
| 75 | + learning_rate = 1e-3 |
| 76 | + optimizer = torch.optim.Adam(q_hat.parameters(), lr=learning_rate) |
| 77 | + scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=100, gamma=0.8) |
| 78 | + num_epochs = 1000 |
| 79 | + |
| 80 | + for epoch in range(num_epochs): |
| 81 | + for i, time in enumerate(train): |
| 82 | + |
| 83 | + res_ode = derivative_back(q_hat, time) - G(q_hat, time) |
| 84 | + res_p_dot = res_ode[:, 0] |
| 85 | + |
| 86 | + loss = res_p_dot.pow(2).mean() |
| 87 | + |
| 88 | + optimizer.zero_grad() |
| 89 | + loss.backward() |
| 90 | + |
| 91 | + optimizer.step() |
| 92 | + scheduler.step() |
| 93 | + |
| 94 | + # Generate test_data and compare to the benchmark |
| 95 | + mu_benchmark = mu_f_array(np.array(test_data), c, g, r, x_0) |
| 96 | + mu_test = np.array(q_hat(test_data)[:, [0]].detach()) |
| 97 | + |
| 98 | + mu_rel_error = np.abs(mu_benchmark - mu_test) / mu_benchmark |
| 99 | + print(f"E(|rel_error(p)|) = {mu_rel_error.mean()}") |
| 100 | + return { |
| 101 | + "t_train": train_data, |
| 102 | + "t_test": test_data, |
| 103 | + "p_test": mu_test, |
| 104 | + "p_benchmark": mu_benchmark, |
| 105 | + "p_rel_error": mu_rel_error, |
| 106 | + "neural_net_solution": q_hat, # interpolator |
| 107 | + } |
| 108 | + |
| 109 | + |
| 110 | +if __name__ == "__main__": |
| 111 | + jsonargparse.CLI(asset_pricing_neural) |
0 commit comments