ในโลกของการลงทุนที่ซับซ้อน การจัดสรรพอร์ตโฟลิโอให้เหมาะสมที่สุดเป็นหนึ่งในความท้าทายที่ยากที่สุด เพราะนักลงทุนต้องสมดุลระหว่างผลตอบแทนสูงสุด ความเสี่ยงต่ำสุด และสภาพคล่องที่เพียงพอ วันนี้ผมจะมาแบ่งปันวิธีการใช้ Multi-Objective Genetic Algorithm (MOGA) ร่วมกับ Large Language Model (LLM) เพื่อสร้างระบบจัดพอร์ตโฟลิโอที่ชาญฉลาด
ปัญหาจริงที่ผมเจอ: Genetic Algorithm ทำงานช้าและได้ผลลัพธ์ไม่ดี
สมัครที่นี่ ผมเคยพัฒนาระบบจัดพอร์ตโฟลิโอด้วย Genetic Algorithm แบบดั้งเดิม โดยใช้ Python ล้วนๆ ปัญหาที่เจอคือ:
# ปัญหา: การคำนวณ Fitness Function ใช้เวลานานมาก
เมื่อมีสินทรัพย์จำนวนมาก ต้องคำนวณทุก Combination
ทำให้ Algorithm ทำงานช้าและหยุดชะงัก
import numpy as np
import random
from typing import List, Tuple
class SlowPortfolioOptimizer:
def __init__(self, assets: List[str], population_size: int = 100):
self.assets = assets
self.population_size = population_size
self.population = []
def create_individual(self) -> np.ndarray:
# สุ่มสัดส่วนการลงทุน
weights = np.random.dirichlet(np.ones(len(self.assets)))
return weights
def calculate_fitness(self, weights: np.ndarray,
returns: np.ndarray,
cov_matrix: np.ndarray) -> float:
"""
คำนวณ Fitness Score - อันนี้คือจุดอ่อน!
ยิ่งมีสินทรัพย์มาก ยิ่งช้า O(n²)
"""
portfolio_return = np.sum(weights * returns)
portfolio_risk = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
# Sharpe Ratio (Simplified)
risk_free_rate = 0.02
sharpe = (portfolio_return - risk_free_rate) / (portfolio_risk + 1e-8)
return sharpe
ปัญหา: ใช้เวลาคำนวณนานเกินไป
optimizer = SlowPortfolioOptimizer(['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'TSLA'])
print("กำลังคำนวณ... รอสักครู่...")จากประสบการณ์ตรง ผมพบว่าวิธีนี้มีข้อจำกัดหลายประการ โดยเฉพาะเมื่อต้องจัดการกับข้อมูลจำนวนมาก ระบบจะทำงานช้าลงอย่างมาก และบางครั้งก็ได้ผลลัพธ์ที่ไม่ใช่ Pareto Optimal
วิธีแก้ไข: ใช้ LLM ช่วยปรับปรุง Genetic Algorithm
ผมค้นพบว่าการนำ LLM มาช่วยในกระบวนการสามารถแก้ปัญหาได้หลายอย่าง:
- การตีความผลลัพธ์: LLM ช่วยวิเคราะห์ว่าทำไม Algorithm เลือกสัดส่วนเฉพาะ
- การกำหนด Mutation Rate แบบ Dynamic: ปรับความน่าจะเป็นในการกลายพันธุ์ตามสภาวะตลาด
- การแนะนำ Constraints: LLM ช่วยเสนอข้อจำกัดที่เหมาะสม
- การอธิบายผลลัพธ์: แปลงตัวเลขซับซ้อนให้เข้าใจง่าย
การตั้งค่า HolySheep AI API
สำหรับการใช้งาน LLM ผมแนะนำให้ใช้ HolySheep AI เพราะมีข้อดีหลายประการ: ราคาประหยัดสูงสุด 85% เมื่อเทียบกับ OpenAI, รองรับ WeChat/Alipay, ความหน่วงต่ำกว่า 50ms และได้เครดิตฟรีเมื่อลงทะเบียน โดยมีราคา 2026 ดังนี้: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok, Gemini 2.5 Flash $2.50/MTok, DeepSeek V3.2 $0.42/MTok
import os
import requests
import json
from typing import Dict, List, Optional
ตั้งค่า HolySheep AI API
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
class HolySheepLLM:
"""ติดต่อ LLM ผ่าน HolySheep AI API"""
def __init__(self, api_key: str = API_KEY):
self.api_key = api_key
self.base_url = BASE_URL
def analyze_portfolio_suggestion(self,
weights: Dict[str, float],
market_conditions: str) -> str:
"""
ใช้ LLM วิเคราะห์และอธิบายข้อเสนอพอร์ตโฟลิโอ
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# สร้าง Prompt สำหรับ LLM
prompt = f"""คุณเป็นผู้เชี่ยวชาญด้านการลงทุน
วิเคราะห์การจัดสรรพอร์ตโฟลิโอต่อไปนี้:
สัดส่วนการลงทุน:
{json.dumps(weights, indent=2)}
สภาวะตลาดปัจจุบัน:
{market_conditions}
โปรดให้คำแนะนำ:
1. ความเสี่ยงของพอร์ตนี้
2. จุดแข็งและจุดอ่อน
3. ข้อเสนอแนะเพื่อปรับปรุง
"""
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "คุณเป็นที่ปรึกษาการลงทุนที่มีประสบการณ์"},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 1000
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
return result['choices'][0]['message']['content']
except requests.exceptions.Timeout:
return "Error: API Timeout - ลองใช้ DeepSeek V3.2 แทนซึ่งเร็วกว่า"
except requests.exceptions.RequestException as e:
return f"Error: {str(e)}"
ทดสอบ
llm_client = HolySheepLLM()
weights = {"AAPL": 0.30, "GOOGL": 0.25, "MSFT": 0.25, "AMZN": 0.20}
market = "ตลาดมีความผันผวนสูง อัตราดอกเบี้ยขาขึ้น"
result = llm_client.analyze_portfolio_suggestion(weights, market)
print(result)ระบบ Multi-Objective Genetic Algorithm สำหรับ Portfolio Optimization
ต่อไปนี้คือโค้ดที่สมบูรณ์สำหรับระบบ MOGA ที่ทำงานร่วมกับ LLM
import numpy as np
import random
from dataclasses import dataclass
from typing import List, Tuple, Dict, Callable
from deap import base, creator, tools, algorithms
import json
@dataclass
class Asset:
name: str
expected_return: float
volatility: float
liquidity: float
sector: str
class MultiObjectivePortfolioGA:
"""
Multi-Objective Genetic Algorithm สำหรับ Portfolio Optimization
รวม LLM เพื่อปรับปรุงคุณภาพผลลัพธ์
"""
def __init__(self,
assets: List[Asset],
llm_client, # HolySheepLLM instance
min_weight: float = 0.01,
max_weight: float = 0.5):
self.assets = assets
self.llm_client = llm_client
self.min_weight = min_weight
self.max_weight = max_weight
self.n_assets = len(assets)
# ตั้งค่า DEAP framework
self._setup_deap()
def _setup_deap(self):
"""ตั้งค่า DEAP สำหรับ Multi-Objective Optimization"""
# ลบ classes เก่าถ้ามี
if hasattr(creator, "FitnessMulti"):
del creator.FitnessMulti
if hasattr(creator, "Individual"):
del creator.Individual
# สร้าง Multi-Objective Fitness (Minimize Risk, Maximize Return)
creator.create("FitnessMulti", base.Fitness, weights=(-1.0, 1.0, 1.0))
creator.create("Individual", list, fitness=creator.FitnessMulti)
self.toolbox = base.Toolbox()
# กำหนดวิธีสร้าง Individual
def create_individual():
# ใช้ Dirichlet distribution เพื่อให้ผลรวม = 1
weights = np.random.dirichlet(np.ones(self.n_assets))
# Clip ให้อยู่ในขอบเขต
weights = np.clip(weights, self.min_weight, self.max_weight)
# Normalize ให้รวม = 1
weights = weights / weights.sum()
return list(weights)
self.toolbox.register("individual", tools.initIterate, creator.Individual, create_individual)
self.toolbox.register("population", tools.initRepeat, list, self.toolbox.individual)
# กำหนด Genetic Operators
self.toolbox.register("evaluate", self._evaluate)
self.toolbox.register("mate", tools.cxSimulatedBinaryBounded,
low=self.min_weight,
up=self.max_weight,
eta=20.0)
self.toolbox.register("mutate", self._adaptive_mutation)
self.toolbox.register("select", tools.selNSGA2)
def _adaptive_mutation(self, individual):
"""Adaptive Mutation - ใช้ LLM ช่วยกำหนดอัตราการกลายพันธุ์"""
# คำนวณอัตราการกลายพันธุ์แบบ Dynamic
fitness_values = [ind.fitness.values for ind in self.toolbox.select(
self.population, k=10)]
# ถ้า fitness ดี ให้ mutation ต่ำ
if fitness_values:
avg_fitness = np.mean([f[1] for f in fitness_values])
mutation_rate = max(0.01, 0.2 - avg_fitness * 0.1)
else:
mutation_rate = 0.1
# Mutate
for i in range(len(individual)):
if random.random() < mutation_rate:
delta = random.gauss(0, 0.1)
individual[i] = max(self.min_weight,
min(self.max_weight,
individual[i] + delta))
# Normalize
total = sum(individual)
individual = [w/total for w in individual]
return individual,
def _evaluate(self, individual: List[float]) -> Tuple[float, float, float]:
"""คำนวณ Fitness: (Risk, Return, Sharpe)"""
weights = np.array(individual)
returns = np.array([a.expected_return for a in self.assets])
volatilities = np.array([a.volatility for a in self.assets])
# คาดหวังผลตอบแทน
expected_return = np.dot(weights, returns)
# ความเสี่ยง (Simplified - ไม่ใช้ Covariance Matrix เต็มรูปแบบ)
risk = np.sqrt(np.sum((weights * volatilities) ** 2))
# สภาพคล่อง (Inverse of liquidity risk)
liquidities = np.array([a.liquidity for a in self.assets])
liquidity_score = np.dot(weights, liquidities)
# Sharpe Ratio (Simplified)
risk_free_rate = 0.02
sharpe = (expected_return - risk_free_rate) / (risk + 1e-8)
# Return fitness values
return risk, expected_return, liquidity_score
def optimize(self,
n_generations: int = 100,
population_size: int = 200,
cxpb: float = 0.9,
mutpb: float = 0.2) -> List[Dict]:
"""
รัน Optimization Process
"""
print(f"เริ่ม Multi-Objective Optimization...")
print(f"จำนวน Generation: {n_generations}")
print(f"ขนาด Population: {population_size}")
# สร้าง Population เริ่มต้น
self.population = self.toolbox.population(n=population_size)
# คำนวณ Fitness เริ่มต้น
fitnesses = [self.toolbox.evaluate(ind) for ind in self.population]
for ind, fit in zip(self.population, fitnesses):
ind.fitness.values = fit
# Evolution Loop
for gen in range(n_generations):
if gen % 10 == 0:
print(f"Generation {gen}/{n_generations} - กำลังประมวลผล...")
# Select next generation
offspring = tools.selNSGA2(self.population, len(self.population))
# Clone individuals
offspring = [self.toolbox.clone(ind) for ind in offspring]
# Crossover
for child1, child2 in zip(offspring[::2], offspring[1::2]):
if random.random() < cxpb:
self.toolbox.mate(child1, child2)
del child1.fitness.values
del child2.fitness.values
# Mutation
for mutant in offspring:
if random.random() < mutpb:
self.toolbox.mutate(mutant)
del mutant.fitness.values
# Evaluate individuals with invalid fitness
invalid_ind = [ind for ind in offspring if not ind.fitness.valid]
fitnesses = [self.toolbox.evaluate(ind) for ind in invalid_ind]
for ind, fit in zip(invalid_ind, fitnesses):
ind.fitness.values = fit
# Select