kjg's blog

import torch
import numpy as np

# 1. 初始化张量
# 直接用列表数据建立
torch.tensor(data) 
# numpy数组转tensor：用numpy数组建立是浅拷贝，如改变np_array，x_np随之改变
x_np = torch.from_numpy(np_array) 
# tensor转numpy数组
t = torch.ones(3)	n = t.numpy()
# 用另一个Tensor建立
torch.ones_like(x_data)
torch.rand_like(x_data, dtype=torch.float)
# 使用随机变量或常量建立
torch.rand(shape)
torch.ones(shape)
torch.zeros(shape)
# 在GPU上创建张量
tensor.to(‘cuda’) 

# 2. 张量的属性
# 形状
tensor.shape
# 数据类型
tensor.dtype 
# 存储设备
tensor.device 

# 3. 张量操作
# 索引和切片
tensor = torch.ones(4, 4)
print('First row: ',tensor[0])
print("First column: ",tensor [:, 0])
print('Last column: ',tensor[: ,-1])
# 拼接 stack会多一个dimension
torch.cat([tensor,tensor1,tensor2],dim=1)
torch.stack([tensor,tensor1,tensor2],dim=1)
# 算数
# 矩阵乘法
tensor @ tensor1 
tensor.matmul(tensor1)
# 元素点乘
tensor * tensor1
tensor.mul(tensor1)
# 聚合 
# 聚合张量的所有值为一个值：tensor.sum()
agg = tensor.sum()
# 把张量转变为python常数值：agg.item()
agg_item = agg.item()  

数据库：cryptoDB
行情表：kline_1m 	分区一：日	分区二：[标的名哈希,10]
# ====================================== 
_date          			date
_symbol            		varchar(30)
_timestamp     			timestamp
_open                 	float8
_high                 	float8
_low                  	float8
_close                	float8
_vol                  	float8
# ======================================

数据库：factorDB
日频因子表：fator_1d		分区一：年  分区二：[标的名哈希,10]
因子列表：[市值 marketcap]
# ======================================
_year          			int
_factor            		varchar(30)
_timestamp     			timestamp 
_symbol            		varchar(30)
_val                  	float8
# ======================================

数据库：factorDB
5分钟频因子表：fator_5m	分区一：日	分区二：[标的名哈希,10]
因子列表：[全局多空比 global_lsr，大户多空比 top_lsr，大户多空持仓比 top_lsr_position，持仓量 open_interest 资费 fundingrate]
# ======================================
_date          			date
_factor            		varchar(30)
_timestamp     			timestamp
_symbol            		varchar(30)
_val                  	float8
# ======================================


数据库：optionDB
表：option_data	分区一：日	分区二：[标的名哈希,25]
# ======================================
_date          			date
_symbol            		varchar(30)
_timestamp     			timestamp
_type            		varchar(10)
_strike_price           float8
_expiration				timestamp
_open_interest			float8
_last_price				float8
_bid_price				float8
_bid_amount				float8
_bid_iv					float8
_ask_price				float8
_ask_amount				float8
_ask_iv					float8
_mark_price				float8
_mark_iv				float8
_underlying_index		varchar(30)
_underlying_price		float8
_delta					float8
_gamma					float8
_vega					float8
_theta					float8
_rho					float8
# ======================================
exchange,symbol,timestamp,local_timestamp,type,strike_price,expiration,open_interest,last_price,bid_price,bid_amount,bid_iv,ask_price,ask_amount,ask_iv,mark_price,mark_iv,underlying_index,underlying_price,delta,gamma,vega,theta,rho
deribit,BTC-27SEP24-160000-P,1704067200027000,1704067200029840,put,160000,1727424000000000,0,2.45,,,,,,,2.4467,75.18,BTC-27SEP24,46593.7,-0.94333,0,45.67483,-6.35109,-1169.77956

import psycopg2
 
# 连接到 PostgreSQL 数据库
conn = psycopg2.connect(database="your_db", user="your_user", password="your_password", host="localhost", port="5432")
cur = conn.cursor()
 
# 定义要插入的数据
data = [('value1', 'value2'), ('value3', 'value4')]
 
try:
    # 开始事务
    cur.execute("BEGIN TRANSACTION;")
    
    # 构建 INSERT INTO 语句并执行
    query = "INSERT INTO target_table (column1, column2) VALUES %s;"
    execute_values(query, data, page_size=1000)
    
    # 提交事务
    cur.execute("COMMIT;")
except Exception as e:
    print("Error occurred during batch insertion: ", str(e))
    # 如果发生错误则回滚事务
    cur.execute("ROLLBACK;")
finally:
    # 关闭数据库连接
    cur.close()
    conn.close()

import numpy as np
import pandas as pd
from gplearn import fitness
# 数据集
from gplearn.genetic import SymbolicRegressor

train_data = pd.read_csv('../data/IC_train.csv', index_col=0, parse_dates=[0])
test_data = pd.read_csv('../data/IC_test.csv', index_col=0, parse_dates=[0])
feature_names = list(train_data.columns)
train_data.loc[:, 'y'] = np.log(train_data['Close'].shift(-4) / train_data['Close'])
train_data.dropna(inplace=True)

from examples.backtest import BackTester

class SymbolicTestor(BackTester):  # 回测的设定
    def init(self):
        self.params = {'factor': pd.Series}

    @BackTester.process_strategy
    def run_(self, *args, **kwargs) -> dict[str: int]:
        factor = np.array(self.params['factor'])
        long_cond = factor > 0
        short_cond = factor < 0
        self.backtest_env['signal'] = np.where(long_cond, 1, np.where(short_cond, -1, np.nan))
        self.construct_position_(keep_raw=True, max_holding_period=1200, take_profit=None, stop_loss=None)

# 回测环境（适应度函数）
comm = [0 / 10000, 0 / 10000]  # 买卖费率
bt = SymbolicTestor(train_data, transact_base='PreClose', commissions=(comm[0], comm[1]))  # 加载数据，根据Close成交,comm是买-卖

def score_func_basic(y, y_pred, sample_weight):  # 因子评价指标
    try:
        _ = bt.run_(factor=y_pred)
        factor_ret = _['annualized_mean']/_['max_drawdown'] if _['max_drawdown'] != 0 else 0 # 可以把max_drawdown换成annualized_std
    except:
        factor_ret = 0
    return factor_ret

def my_gplearn(function_set, my_fitness, pop_num=100, gen_num=3, tour_num=10, random_state = 42, feature_names=None):
    # pop_num, gen_num, tour_num的几个可选值：500, 5, 50; 1000, 3, 20; 1000, 15, 100
    metric = fitness.make_fitness(function=my_fitness, # function(y, y_pred, sample_weight) that returns a floating point number.
                        greater_is_better=True,  # 上述y是输入的目标y向量，y_pred是genetic program中的预测值，sample_weight是样本权重向量
                        wrap=False)  # 不保存，运行的更快 # gplearn.fitness.make_fitness(function, greater_is_better, wrap=True)
    return SymbolicRegressor(population_size=pop_num,  # 每一代公式群体中的公式数量 500，100
                              generations=gen_num,  # 公式进化的世代数量 10，3
                              metric=metric,  # 适应度指标，这里是前述定义的通过 大于0做多，小于0做空的 累积净值/最大回撤 的评判函数
                              tournament_size=tour_num,  # 在每一代公式中选中tournament的规模，对适应度最高的公式进行变异或繁殖 50
                              function_set=function_set,
                              const_range=(-1.0, 1.0),  # 公式中包含的常数范围
                              parsimony_coefficient='auto',
                              # 对较大树的惩罚,默认0.001，auto则用c = Cov(l,f)/Var( l), where Cov(l,f) is the covariance between program size l and program fitness f in the population, and Var(l) is the variance of program sizes.
                              stopping_criteria=100.0,  # 是对metric的限制（此处为收益/回撤）
                              init_depth=(2, 3),  # 公式树的初始化深度，树深度最小2层，最大6层
                              init_method='half and half',  # 树的形状，grow生分枝整的不对称，full长出浓密
                              p_crossover=0.8,  # 交叉变异概率 0.8
                              p_subtree_mutation=0.05,  # 子树变异概率
                              p_hoist_mutation=0.05,  # hoist变异概率 0.15
                              p_point_mutation=0.05,  # 点变异概率
                              p_point_replace=0.05,  # 点变异中每个节点进行变异进化的概率
                              max_samples=1.0,  # The fraction of samples to draw from X to evaluate each program on.
                              feature_names=feature_names, warm_start=False, low_memory=False,
                              n_jobs=1,
                              verbose=1,
                              random_state=random_state)

# 函数集
function_set=['add', 'sub', 'mul', 'div', 'sqrt', 'log',  # 用于构建和进化公式使用的函数集
                    'abs', 'neg', 'inv', 'sin', 'cos', 'tan', 'max', 'min',
                    # 'if', 'gtpn', 'andpn', 'orpn', 'ltpn', 'gtp', 'andp', 'orp', 'ltp', 'gtn', 'andn', 'orn', 'ltn', 'delayy', 'delta', 'signedpower', 'decayl', 'stdd', 'rankk'
                    ] # 最后一行是自己的函数，目前不用自己函数效果更好

my_cmodel_gp = my_gplearn(function_set, score_func_basic, random_state=0,
                          feature_names=feature_names)  # 可以通过换random_state来生成不同因子
my_cmodel_gp.fit(train_data.loc[:, :'rank_num'].values, train_data.loc[:, 'y'].values)
print(my_cmodel_gp)

# 策略结果
factor = my_cmodel_gp.predict(test_data.values)
bt_test = SymbolicTestor(test_data, transact_base='PreClose', commissions=(comm[0], comm[1]))  # 加载数据，根据Close成交,comm是买-卖
bt_test.run_(factor=factor)
md = bt_test.summary()
print(md.out_stats)
print(bt.fees_factor)
md.plot_(comm=comm, show_bool=True)