Pancake's Personal Website

import geatpy as ea
import numpy as np


# 示例说明：
# a = [0~30连续数] * _num
# b = [1 ~ num num个数排列排序] 即数列 1~num 不重复排列
# c = [0~30离散数（整数）] * _num
def func(a, b, c):
    return np.sin(np.pi * a * b) * np.cos(np.pi * c * b) * np.tan(np.pi * a * b * c)


# 初始化文件num大小
_num = 10


# 自定义 GA
class My_nsga(ea.Problem):
    def __init__(self):
        global _num
        name = 'GEATPY Facilitate Learning'
        # 只优化func一个变量
        M = 1
        # 最大化变量
        maxormins = [-1] * M
        # 一共有 a.size + b.size + c.size 个数
        Dim = _num + _num + _num
        # a 连续， b and c 离散
        varTypes = [0] * _num + [1] * _num + [1] * _num
        # a ∈ [0,30],  b ∈ [1,_num], c ∈ [0,30]
        lb = [0] * _num + [1] * _num + [0] * _num
        ub = [30] * _num + [_num] * _num + [30] * _num
        # 下限可取
        lbin = [1] * Dim
        # 上限可取
        ubin = [1] * Dim
        # 保存最大数据
        self.max_ans = 0
        # epoch
        self.epoch = 0
        # super
        ea.Problem.__init__(self, name, M, maxormins, Dim, varTypes, lb, ub, lbin, ubin)

    # 目标函数即神经网络返回值
    def evalVars(self, Vars):
        # init ans
        ans = np.zeros(len(Vars), dtype=float).reshape(len(Vars), 1)
        for i in range(len(Vars)):
            # 一定要确定区域选择是否正确
            a = Vars[i][:_num]
            # 可以选择int类型防止特定错误，因为正常输出时为float，具体问题可能会出错
            b = list(map(int, Vars[i][_num: _num + _num]))
            c = Vars[i][-_num:]
            score = func(a, b, c)
            ans[i] = score.max()
        self.max_ans = (now_max := ans.max())
        print(f"Epoch: {self.epoch + 1}, Epoch Max: {now_max}, Global Max: {self.max_ans}")
        self.epoch += 1
        return ans


# 运行 GA
def Run_nsga(ndind=10, maxgen=100, *args, **kwargs):
    global _num
    # init problem
    problem = My_nsga()
    # 染色体编码方式为[格雷编码，排列编码，格雷编码]
    encodings = ["RI", "P", "RI"]
    # 设置具体field，一定要多加这一部，geatpy还没有具体到能直接输入encoding: List[str]，需要进行转换才能使用
    # 一定要注意数组范围，报错的话一定要仔细检查！！！
    field1 = ea.crtfld(encodings[0], problem.varTypes[:_num], problem.ranges[:, :_num], problem.borders[:, :_num])
    field2 = ea.crtfld(encodings[1], problem.varTypes[_num: _num + _num], problem.ranges[:, _num: _num + _num], problem.borders[:, _num: _num + _num])
    field3 = ea.crtfld(encodings[2], problem.varTypes[-_num:], problem.ranges[:, -_num:], problem.borders[:, -_num:])
    # 合并field
    fields = [field1, field2, field3]
    # 代入模板
    myAlgorithm = ea.soea_psy_EGA_templet(problem, ea.PsyPopulation(Encodings=encodings, Fields=fields, NIND=ndind), MAXGEN=maxgen, logTras=0)
    # 是否画图
    myAlgorithm.drawing = 0
    # 进行优化
    res = ea.optimize(myAlgorithm, seed=1, verbose=False, drawing=0, outputMsg=True, drawLog=False, saveFlag=False, dirName='result')
    return res['Vars'][0]


if __name__ == "__main__":
    Run_nsga(50, 1000)

import geatpy as ea

import numpy as np

# 示例说明：

# a = [0~30连续数] * _num

# b = [1 ~ num num个数排列排序] 即数列 1~num 不重复排列

# c = [0~30离散数（整数）] * _num

def func(a, b, c):

return np.sin(np.pi * a * b) * np.cos(np.pi * c * b) * np.tan(np.pi * a * b * c)

# 初始化文件num大小

_num = 10

# 自定义 GA

class My_nsga(ea.Problem):

def __init__(self):

global _num

name = 'GEATPY Facilitate Learning'

# 只优化func一个变量

M = 1

# 最大化变量

maxormins = [-1] * M

# 一共有 a.size + b.size + c.size 个数

Dim = _num + _num + _num

# a 连续， b and c 离散

varTypes = [0] * _num + [1] * _num + [1] * _num

# a ∈ [0,30], b ∈ [1,_num], c ∈ [0,30]

lb = [0] * _num + [1] * _num + [0] * _num

ub = [30] * _num + [_num] * _num + [30] * _num

# 下限可取

lbin = [1] * Dim

# 上限可取

ubin = [1] * Dim

# 保存最大数据

self.max_ans = 0

# epoch

self.epoch = 0

# super

ea.Problem.__init__(self, name, M, maxormins, Dim, varTypes, lb, ub, lbin, ubin)

# 目标函数即神经网络返回值

def evalVars(self, Vars):

# init ans

ans = np.zeros(len(Vars), dtype=float).reshape(len(Vars), 1)

for i in range(len(Vars)):

# 一定要确定区域选择是否正确

a = Vars[i][:_num]

# 可以选择int类型防止特定错误，因为正常输出时为float，具体问题可能会出错

b = list(map(int, Vars[i][_num: _num + _num]))

c = Vars[i][-_num:]

score = func(a, b, c)

ans[i] = score.max()

self.max_ans = (now_max := ans.max())

print(f"Epoch: {self.epoch + 1}, Epoch Max: {now_max}, Global Max: {self.max_ans}")

self.epoch += 1

return ans

# 运行 GA

def Run_nsga(ndind=10, maxgen=100, *args, **kwargs):

global _num

# init problem

problem = My_nsga()

# 染色体编码方式为[格雷编码，排列编码，格雷编码]

encodings = ["RI", "P", "RI"]

# 设置具体field，一定要多加这一部，geatpy还没有具体到能直接输入encoding: List[str]，需要进行转换才能使用

# 一定要注意数组范围，报错的话一定要仔细检查！！！

field1 = ea.crtfld(encodings[0], problem.varTypes[:_num], problem.ranges[:, :_num], problem.borders[:, :_num])

field2 = ea.crtfld(encodings[1], problem.varTypes[_num: _num + _num], problem.ranges[:, _num: _num + _num], problem.borders[:, _num: _num + _num])

field3 = ea.crtfld(encodings[2], problem.varTypes[-_num:], problem.ranges[:, -_num:], problem.borders[:, -_num:])

# 合并field

fields = [field1, field2, field3]

# 代入模板

myAlgorithm = ea.soea_psy_EGA_templet(problem, ea.PsyPopulation(Encodings=encodings, Fields=fields, NIND=ndind), MAXGEN=maxgen, logTras=0)

# 是否画图

myAlgorithm.drawing = 0

# 进行优化

res = ea.optimize(myAlgorithm, seed=1, verbose=False, drawing=0, outputMsg=True, drawLog=False, saveFlag=False, dirName='result')

return res['Vars'][0]

if __name__ == "__main__":

Run_nsga(50, 1000)

Python 深入学习

生成器与迭代器

# 迭代器
class Fibonacci:
    def __init__(self, n):
        self.a = 0
        self.b = 1
        self.count = 0
        self.n = n

    def __iter__(self):
        return self

    def __next__(self):
        if self.count > self.n:
            raise StopIteration
        now = self.a
        self.a, self.b = self.b, self.a + self.b
        self.count += 1
        return now


# 生成器
def fibonacci(n):
    a, b, counter = 0, 1, 0
    while True:
        if counter > n:
            return
        yield a
        a, b = b, a + b
        counter += 1


# 迭代器
f = Fibonacci(10)

while True:
    try:
        print(next(f), end=" ")
    except StopIteration:
        print('\n')
        break

# 生成器
f = fibonacci(0)

while True:
    try:
        print(f.__next__(), end=" ")
    except StopIteration:
        break

# 迭代器

class Fibonacci:

def __init__(self, n):

self.a = 0

self.b = 1

self.count = 0

self.n = n

def __iter__(self):

return self

def __next__(self):

if self.count > self.n:

raise StopIteration

now = self.a

self.a, self.b = self.b, self.a + self.b

self.count += 1

return now

# 生成器

def fibonacci(n):

a, b, counter = 0, 1, 0

while True:

if counter > n:

return

yield a

a, b = b, a + b

counter += 1

# 迭代器

f = Fibonacci(10)

while True:

try:

print(next(f), end=" ")

except StopIteration:

print('\n')

break

# 生成器

f = fibonacci(0)

while True:

try:

print(f.__next__(), end=" ")

except StopIteration:

break

生成器进阶 - send

from itertools import count


def func1():
    a = yield 1
    print(a)
    b = yield 2
    print(b)
    c = yield 3
    print(c)
    yield "done"


func = func1()
for i in count():
    try:
        print(f"interaiton setp : {i},", f"answer = {func.__next__()}")
    except StopIteration:
        break

func = func1()
print(func.__next__())
print(func.send(4))
print(func.send(5))
print(func.send(6))
func.close()

from itertools import count

def func1():

a = yield 1

print(a)

b = yield 2

print(b)

c = yield 3

print(c)

yield "done"

func = func1()

for i in count():

try:

print(f"interaiton setp : {i},", f"answer = {func.__next__()}")

except StopIteration:

break

func = func1()

print(func.__next__())

print(func.send(4))

print(func.send(5))

print(func.send(6))

func.close()

自定义错误类型

class NameTooShortError(Exception):
    pass


def validate(name):
    if len(name) < 10:
        raise NameTooShortError(name)


validate("321")

class NameTooShortError(Exception):

pass

def validate(name):

if len(name) < 10:

raise NameTooShortError(name)

validate("321")

修饰器

def decorate_1(func):
    def inner1(*args):
        print('1', func.__name__, args)
        func(*args)

    return inner1


def decorate_2(func):
    def inner2(*args):
        print('2', func.__name__, args)
        func(*args)

    return inner2


@decorate_1
@decorate_2
def main(*args):
    print('test', args)


main(1, 2, 3, 4)

from functools import wraps


def wrapper(func):
    @wraps(func)
    def inner_function():
        pass

    return inner_function


@wrapper
def wrapped():
    pass


print(wrapped.__name__)

def decorate_1(func):

def inner1(*args):

print('1', func.__name__, args)

func(*args)

return inner1

def decorate_2(func):

def inner2(*args):

print('2', func.__name__, args)

func(*args)

return inner2

@decorate_1

@decorate_2

def main(*args):

print('test', args)

main(1, 2, 3, 4)

from functools import wraps

def wrapper(func):

@wraps(func)

def inner_function():

pass

return inner_function

@wrapper

def wrapped():

pass

print(wrapped.__name__)

import time
from functools import wraps


def cal_time(unit='s'):
    def input_func(func):
        units = ['ms', 's', 'min', 'hour', 'day']
        times = [1000, 1, 1 / 60, 1 / 3600, 1 / 3600 / 24]
        if unit.lower() not in units:
            raise TypeError(unit)
        time_process = times[units.index(unit)]

        @wraps(func)
        def wrap(*args, **kwargs):
            begin_time = time.time()
            ans = func(*args, **kwargs)
            print(func.__name__, f"用时 {(time.time() - begin_time) * time_process}")
            return ans

        return wrap

    return input_func

import time

from functools import wraps

def cal_time(unit='s'):

def input_func(func):

units = ['ms', 's', 'min', 'hour', 'day']

times = [1000, 1, 1 / 60, 1 / 3600, 1 / 3600 / 24]

if unit.lower() not in units:

raise TypeError(unit)

time_process = times[units.index(unit)]

@wraps(func)

def wrap(*args, **kwargs):

begin_time = time.time()

ans = func(*args, **kwargs)

print(func.__name__, f"用时 {(time.time() - begin_time) * time_process}")

return ans

return wrap

return input_func

常见修饰器

# 无需实例化对象，无需self，需要cls
@classmethod
# 无需实例化对象，无需self和cls
@staticmethod
import functools
# 缓存结果，算法上可用
@functools.cache
# 自定义修饰器时使用
@functools.wraps
# 特定代码替换解释器加快代码运行速度
from numba import jit
@jit

# 无需实例化对象，无需self，需要cls

@classmethod

# 无需实例化对象，无需self和cls

@staticmethod

import functools

# 缓存结果，算法上可用

@functools.cache

# 自定义修饰器时使用

@functools.wraps

# 特定代码替换解释器加快代码运行速度

from numba import jit

@jit

海象运算符:=（python>=3.8）

# 简单来说就是简化代码，使用更方便，需要注意运算优先级
# 下列代码需要加括号才能使 n = len(a)
if (n := len(a)) > 10:
    print(f"List is too long ({n} elements, expected <= 10)")

# 简单来说就是简化代码，使用更方便，需要注意运算优先级

# 下列代码需要加括号才能使 n = len(a)

if (n := len(a)) > 10:

print(f"List is too long ({n} elements, expected <= 10)")

特殊的list deque

因为最近在写算法题，用的是C++，看了下python也有deque

from collections import deque
d=deque()

1 2	from collections import deque d=deque()

拓扑算法库

from graphlib import TopologicalSorter

1	from graphlib import TopologicalSorter

collections库

"""
https://blog.csdn.net/StarSky_Ye/article/details/119272313
"""

"""

https://blog.csdn.net/StarSky_Ye/article/details/119272313

"""

下划线(_)的作用

# codeTest.py
_a = "_a"
a = "a"

def _test_underline():
    print("_test_underline")

def test_underline():
    print("test_underline")

# code.py
from codeTest import *

test_underline()
# _test_underline()
# ↑报错
print(a)
# print(_a)
# ↑报错

# codeTest.py

_a = "_a"

a = "a"

def _test_underline():

print("_test_underline")

def test_underline():

print("test_underline")

# code.py

from codeTest import *

test_underline()

# _test_underline()

# ↑报错

print(a)

# print(_a)

# ↑报错

# 初始化数组
matrix = [0 for _ in range(10)]

1 2	# 初始化数组 matrix = [0 for _ in range(10)]

class A:
    __private_var = 1000
 
    def __init__(self) -> None:
        self.__private_var2 = 2000
        self.var3 = 3000
        self.__print("A.__print")

    def __print(self, x):
        print(x)

a = A()
print(a.var3)
# print(a.__private_var2)
# ↑报错
# print(a.__print("x"))
# ↑报错

class A:

__private_var = 1000

def __init__(self) -> None:

self.__private_var2 = 2000

self.var3 = 3000

self.__print("A.__print")

def __print(self, x):

print(x)

a = A()

print(a.var3)

# print(a.__private_var2)

# ↑报错

# print(a.__print("x"))

# ↑报错

正则表达式

这东西自己看的时候可谓是眼花缭乱，看到一半我就不想看了，因为本人没用过，看b站的时候刷到了，推荐一下大佬的学习网站

https://r2coding.com/ 里面还有很多计算机学习资料有兴趣的可以看看

遗传算法背包问题

排列编码

import pandas as pd
import geatpy as ea
import numpy as np
import random
from multiprocessing import Pool


class Data:
    market, market_value = None, None

    def __init__(self, df, platform, platform_max):
        self.platform = platform
        self.platform_max = platform_max
        self.x = df['交易量']
        self.y = df["交易费用"]

    def getResult(self, choose, average):
        choose_dict = [{"index": choose[i], "x": self.x[i], "y": self.y[i]} for i in range(len(choose))]
        choose_dict = sorted(choose_dict, key=lambda x: x["index"])
        result = float("inf")
        for i in range(self.platform):
            market_value = 0
            market = 0
            for j in range(average):
                if choose_dict[i * average + j]["x"] + market > self.platform_max:
                    result = min(result, market_value)
                    break
                else:
                    market_value += choose_dict[i * average + j]["y"]
                    market += choose_dict[i * average + j]["x"]
            if market_value <= self.platform_max:
                result = min(result, market_value)
        return result


# GA 参数设置
class My_nsga(ea.Problem):
    def __init__(self, dim, df, platform, platform_max):
        name = 'GA-NET'
        M = 1
        maxormins = [-1] * M
        Dim = dim
        varTypes = [1] * Dim
        lb = [0] * Dim
        ub = [Dim - 1] * Dim
        lbin = [1] * Dim
        ubin = [1] * Dim
        self.count = 1
        self.data = Data(df, platform, platform_max)
        self.average = int(len(df) / platform)

        ea.Problem.__init__(self, name, M, maxormins, Dim, varTypes, lb, ub, lbin, ubin)

    # 目标函数即神经网络返回值
    def evalVars(self, Vars):
        ans = np.zeros(len(Vars), dtype=float).reshape(len(Vars), 1)
        for i in range(len(Vars)):
            ans[i][0] = self.data.getResult(Vars[i], self.average)
        # print(ans.max(axis=0))
        return ans


# 运行 GA
def Run_nsga(dim, df, platform, platform_max, loop_id, ndind=30, maxgen=1500):
    problem = My_nsga(dim, df, platform, platform_max)
    myAlgorithm = ea.soea_EGA_templet(problem, ea.Population(Encoding='P', NIND=ndind), MAXGEN=maxgen, logTras=0)
    # 要得到图把 drawing = 1
    saveFlag = 1
    res = ea.optimize(myAlgorithm, seed=1, verbose=False, drawing=1, outputMsg=1, drawLog=1, saveFlag=saveFlag, dirName='result{}'.format(loop_id))
    if saveFlag:
        np.savetxt("./result{}/result.csv".format(i), res['Vars'][0], delimiter=",")
    return res['ObjV'][0][0]


if __name__ == "__main__":
    df = pd.read_excel("data.xlsx", sheet_name=None, engine='openpyxl')
    # 设置每张表分成的 platform 数量
    platforms = [2, 4, 10, 20, 10, 50, 50, 100]
    # 设置每张表 platform 最大值
    platform_maxs = [2000, 1000, 1500, 800, 500, 100, 416, 208]
    # 设置是否训练表
    is_train = [0, 0, 1, 1, 0, 0, 0, 0]
    values = list(df.values())[:4]
    pool = Pool(sum(is_train))
    for i in range(len(values)):
        if not is_train[i]:
            continue
        # 最后两个参数：种群数量，最大迭代次数
        pool.apply_async(Run_nsga, args=(len(values[i]), values[i], platforms[i], platform_maxs[i], i, 500, 600))
    pool.close()
    pool.join()

import pandas as pd

import geatpy as ea

import numpy as np

import random

from multiprocessing import Pool

class Data:

market, market_value = None, None

def __init__(self, df, platform, platform_max):

self.platform = platform

self.platform_max = platform_max

self.x = df['交易量']

self.y = df["交易费用"]

def getResult(self, choose, average):

choose_dict = [{"index": choose[i], "x": self.x[i], "y": self.y[i]} for i in range(len(choose))]

choose_dict = sorted(choose_dict, key=lambda x: x["index"])

result = float("inf")

for i in range(self.platform):

market_value = 0

market = 0

for j in range(average):

if choose_dict[i * average + j]["x"] + market > self.platform_max:

result = min(result, market_value)

break

else:

market_value += choose_dict[i * average + j]["y"]

market += choose_dict[i * average + j]["x"]

if market_value <= self.platform_max:

result = min(result, market_value)

return result

# GA 参数设置

class My_nsga(ea.Problem):

def __init__(self, dim, df, platform, platform_max):

name = 'GA-NET'

M = 1

maxormins = [-1] * M

Dim = dim

varTypes = [1] * Dim

lb = [0] * Dim

ub = [Dim - 1] * Dim

lbin = [1] * Dim

ubin = [1] * Dim

self.count = 1

self.data = Data(df, platform, platform_max)

self.average = int(len(df) / platform)

ea.Problem.__init__(self, name, M, maxormins, Dim, varTypes, lb, ub, lbin, ubin)

# 目标函数即神经网络返回值

def evalVars(self, Vars):

ans = np.zeros(len(Vars), dtype=float).reshape(len(Vars), 1)

for i in range(len(Vars)):

ans[i][0] = self.data.getResult(Vars[i], self.average)

# print(ans.max(axis=0))

return ans

# 运行 GA

def Run_nsga(dim, df, platform, platform_max, loop_id, ndind=30, maxgen=1500):

problem = My_nsga(dim, df, platform, platform_max)

myAlgorithm = ea.soea_EGA_templet(problem, ea.Population(Encoding='P', NIND=ndind), MAXGEN=maxgen, logTras=0)

# 要得到图把 drawing = 1

saveFlag = 1

res = ea.optimize(myAlgorithm, seed=1, verbose=False, drawing=1, outputMsg=1, drawLog=1, saveFlag=saveFlag, dirName='result{}'.format(loop_id))

if saveFlag:

np.savetxt("./result{}/result.csv".format(i), res['Vars'][0], delimiter=",")

return res['ObjV'][0][0]

if __name__ == "__main__":

df = pd.read_excel("data.xlsx", sheet_name=None, engine='openpyxl')

# 设置每张表分成的 platform 数量

platforms = [2, 4, 10, 20, 10, 50, 50, 100]

# 设置每张表 platform 最大值

platform_maxs = [2000, 1000, 1500, 800, 500, 100, 416, 208]

# 设置是否训练表

is_train = [0, 0, 1, 1, 0, 0, 0, 0]

values = list(df.values())[:4]

pool = Pool(sum(is_train))

for i in range(len(values)):

if not is_train[i]:

continue

# 最后两个参数：种群数量，最大迭代次数

pool.apply_async(Run_nsga, args=(len(values[i]), values[i], platforms[i], platform_maxs[i], i, 500, 600))

pool.close()

pool.join()

格雷编码（选择编码）

import pandas as pd
import geatpy as ea
import numpy as np
import random
from multiprocessing import Pool


class Data:
    market, market_value = None, None

    def __init__(self, df, platform, platform_max):
        self.platform = platform
        self.platform_max = platform_max
        self.x = df['交易量']
        self.y = df["交易费用"]

    def load(self, choose):
        self.market = [[] for _ in range(self.platform)]
        self.market_value = [[] for _ in range(self.platform)]
        for i in range(len(choose)):
            if choose[i] != 0:
                self.market[choose[i] - 1].append(self.x[i])
                self.market_value[choose[i] - 1].append(self.y[i])

    def getResult(self):
        if max(list(map(lambda x: sum(x), self.market))) > self.platform_max:
            return sum(list(map(lambda x: 0 if sum(x) < self.platform_max else self.platform_max - sum(x), self.market)))
        else:
            return min(list(map(lambda x: sum(x), self.market_value)))


# GA 参数设置
class My_nsga(ea.Problem):
    def __init__(self, dim, df, platform, platform_max):
        name = 'GA-NET'
        M = 1
        maxormins = [-1] * M
        Dim = dim
        varTypes = [1] * Dim
        lb = [0] * Dim
        ub = [platform] * Dim
        lbin = [1] * Dim
        ubin = [1] * Dim
        self.count = 1
        self.data = Data(df, platform, platform_max)

        ea.Problem.__init__(self, name, M, maxormins, Dim, varTypes, lb, ub, lbin, ubin)

    # 目标函数即神经网络返回值
    def evalVars(self, Vars):
        ans = np.zeros(len(Vars), dtype=float).reshape(len(Vars), 1)
        for i in range(len(Vars)):
            self.data.load(Vars[i])
            ans[i][0] = self.data.getResult()
        # print(ans.max(axis=0))
        return ans


# 得到初始化种群方式1
def getProphet(dim, ndind, df, platform, platform_max):
    prophet = np.zeros(shape=(ndind, dim), dtype=int)
    x = df['交易量']
    for i in range(ndind):
        # 使得platform尽可能分配均匀
        platforms = [i for i in range(platform + 1)]
        now = [0 for _ in range(platform)]
        for j in range(i % dim, dim):
            if not platforms:
                platforms = [i for i in range(platform + 1)]
            choice = random.choice(platforms)
            platforms.remove(choice)
            if choice != 0:
                now[choice - 1] += x[j]
                if now[choice - 1] < platform_max:
                    prophet[i][j] = choice
        for j in range(0, i % dim):
            if not platforms:
                platforms = [i for i in range(platform + 1)]
            choice = random.choice(platforms)
            platforms.remove(choice)
            if choice != 0:
                now[choice - 1] += x[j]
                if now[choice - 1] < platform_max:
                    prophet[i][j] = choice
    return prophet


# 得到初始化种群方式2
def getProphet2(dim, ndind, df, platform, platform_max):
    prophet = np.zeros(shape=(ndind, dim), dtype=int)
    x = df['交易量']
    for i in range(ndind):
        # 真随机抽取初始化
        now = [0 for _ in range(platform)]
        for j in range(i % dim, dim):
            choice = random.choice([i for i in range(platform + 1)])
            if choice != 0:
                now[choice - 1] += x[j]
                if now[choice - 1] < platform_max:
                    prophet[i][j] = choice
        for j in range(0, i % dim):
            choice = random.choice([i for i in range(platform + 1)])
            if choice != 0:
                now[choice - 1] += x[j]
                if now[choice - 1] < platform_max:
                    prophet[i][j] = choice
    return prophet


# 运行 GA
def Run_nsga(dim, df, platform, platform_max, loop_id, ndind=30, maxgen=1500):
    problem = My_nsga(dim, df, platform, platform_max)
    myAlgorithm = ea.soea_EGA_templet(problem, ea.Population(Encoding='RI', NIND=ndind), MAXGEN=maxgen, logTras=0)
    prophet = getProphet(dim, ndind, df, platform, platform_max)
    # 要得到图把 drawing = 1
    saveFlag = 1
    res = ea.optimize(myAlgorithm, prophet=prophet, seed=1, verbose=False, drawing=1, outputMsg=1, drawLog=1, saveFlag=saveFlag, dirName='result{}'.format(loop_id))
    if saveFlag:
        np.savetxt("./result{}/result.csv".format(i), res['Vars'][0], delimiter=",")
    return res['ObjV'][0][0]


if __name__ == "__main__":
    df = pd.read_excel("data.xlsx", sheet_name=None, engine='openpyxl')
    # 设置每张表分成的 platform 数量
    platforms = [2, 4, 10, 20, 10, 50, 50, 100]
    # 设置每张表 platform 最大值
    platform_maxs = [2000, 1000, 1500, 800, 500, 100, 416, 208]
    # platform_maxs[4] = [100, 200, 300, 400, 500, 500, 600, 700, 800, 900]
    # 设置是否训练表
    # is_train = [1, 1, 1, 1, 1, 1, 1, 1]
    is_train = [1, 1, 1, 1, 0, 0, 0, 0]
    result = []
    result_value = []
    values = list(df.values())[:8]
    pool = Pool(sum(is_train))
    for i in range(len(values)):
        if not is_train[i]:
            continue
        # 最后两个参数：种群数量，最大迭代次数
        pool.apply_async(Run_nsga, args=(len(values[i]), values[i], platforms[i], platform_maxs[i], i, 500, 1000))
    pool.close()
    pool.join()
    #     now_value = Run_nsga(len(values[i]), values[i], platforms[i], platform_maxs[i], i, 500, 1000)
    #     result_value.append(now_value)
    # np.savetxt("result_values.csv", result_value, delimiter=",")

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

import pandas as pd

import geatpy as ea

import numpy as np

import random

from multiprocessing import Pool

class Data:

market, market_value = None, None

def __init__(self, df, platform, platform_max):

self.platform = platform

self.platform_max = platform_max

self.x = df['交易量']

self.y = df["交易费用"]

def load(self, choose):

self.market = [[] for _ in range(self.platform)]

self.market_value = [[] for _ in range(self.platform)]

for i in range(len(choose)):

if choose[i] != 0:

self.market[choose[i] - 1].append(self.x[i])

self.market_value[choose[i] - 1].append(self.y[i])

def getResult(self):

if max(list(map(lambda x: sum(x), self.market))) > self.platform_max:

return sum(list(map(lambda x: 0 if sum(x) < self.platform_max else self.platform_max - sum(x), self.market)))

else:

return min(list(map(lambda x: sum(x), self.market_value)))

# GA 参数设置

class My_nsga(ea.Problem):

def __init__(self, dim, df, platform, platform_max):

name = 'GA-NET'

M = 1

maxormins = [-1] * M

Dim = dim

varTypes = [1] * Dim

lb = [0] * Dim

ub = [platform] * Dim

lbin = [1] * Dim

ubin = [1] * Dim

self.count = 1

self.data = Data(df, platform, platform_max)

ea.Problem.__init__(self, name, M, maxormins, Dim, varTypes, lb, ub, lbin, ubin)

# 目标函数即神经网络返回值

def evalVars(self, Vars):

ans = np.zeros(len(Vars), dtype=float).reshape(len(Vars), 1)

for i in range(len(Vars)):

self.data.load(Vars[i])

ans[i][0] = self.data.getResult()

# print(ans.max(axis=0))

return ans

# 得到初始化种群方式1

def getProphet(dim, ndind, df, platform, platform_max):

prophet = np.zeros(shape=(ndind, dim), dtype=int)

x = df['交易量']

for i in range(ndind):

# 使得platform尽可能分配均匀

platforms = [i for i in range(platform + 1)]

now = [0 for _ in range(platform)]

for j in range(i % dim, dim):

if not platforms:

platforms = [i for i in range(platform + 1)]

choice = random.choice(platforms)

platforms.remove(choice)

if choice != 0:

now[choice - 1] += x[j]

if now[choice - 1] < platform_max:

prophet[i][j] = choice

for j in range(0, i % dim):

if not platforms:

platforms = [i for i in range(platform + 1)]

choice = random.choice(platforms)

platforms.remove(choice)

if choice != 0:

now[choice - 1] += x[j]

if now[choice - 1] < platform_max:

prophet[i][j] = choice

return prophet

# 得到初始化种群方式2

def getProphet2(dim, ndind, df, platform, platform_max):

prophet = np.zeros(shape=(ndind, dim), dtype=int)

x = df['交易量']

for i in range(ndind):

# 真随机抽取初始化

now = [0 for _ in range(platform)]

for j in range(i % dim, dim):

choice = random.choice([i for i in range(platform + 1)])

if choice != 0:

now[choice - 1] += x[j]

if now[choice - 1] < platform_max:

prophet[i][j] = choice

for j in range(0, i % dim):

choice = random.choice([i for i in range(platform + 1)])

if choice != 0:

now[choice - 1] += x[j]

if now[choice - 1] < platform_max:

prophet[i][j] = choice

return prophet

# 运行 GA

def Run_nsga(dim, df, platform, platform_max, loop_id, ndind=30, maxgen=1500):

problem = My_nsga(dim, df, platform, platform_max)

myAlgorithm = ea.soea_EGA_templet(problem, ea.Population(Encoding='RI', NIND=ndind), MAXGEN=maxgen, logTras=0)

prophet = getProphet(dim, ndind, df, platform, platform_max)

# 要得到图把 drawing = 1

saveFlag = 1

res = ea.optimize(myAlgorithm, prophet=prophet, seed=1, verbose=False, drawing=1, outputMsg=1, drawLog=1, saveFlag=saveFlag, dirName='result{}'.format(loop_id))

if saveFlag:

np.savetxt("./result{}/result.csv".format(i), res['Vars'][0], delimiter=",")

return res['ObjV'][0][0]

if __name__ == "__main__":

df = pd.read_excel("data.xlsx", sheet_name=None, engine='openpyxl')

# 设置每张表分成的 platform 数量

platforms = [2, 4, 10, 20, 10, 50, 50, 100]

# 设置每张表 platform 最大值

platform_maxs = [2000, 1000, 1500, 800, 500, 100, 416, 208]

# platform_maxs[4] = [100, 200, 300, 400, 500, 500, 600, 700, 800, 900]

# 设置是否训练表

# is_train = [1, 1, 1, 1, 1, 1, 1, 1]

is_train = [1, 1, 1, 1, 0, 0, 0, 0]

result = []

result_value = []

values = list(df.values())[:8]

pool = Pool(sum(is_train))

for i in range(len(values)):

if not is_train[i]:

continue

# 最后两个参数：种群数量，最大迭代次数

pool.apply_async(Run_nsga, args=(len(values[i]), values[i], platforms[i], platform_maxs[i], i, 500, 1000))

pool.close()

pool.join()

# now_value = Run_nsga(len(values[i]), values[i], platforms[i], platform_maxs[i], i, 500, 1000)

# result_value.append(now_value)

# np.savetxt("result_values.csv", result_value, delimiter=",")

两个编码方式有差异，我个人认为排列编码会比较好一点，逻辑更加高一些。代码中为EGA模板（即带有精英选择机制的遗传算法），测试下来可能NSGA-ii模板会更优一些，可以在测试的时候选择你想要的算法及结果。data.xlsx