gyz/TurtleTrade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ai更新架构

Browse Source
This commit is contained in:
guoyz 2025-05-20 22:54:54 +08:00
parent 195341b70c
commit fb18ccc986
8 changed files with 220 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

206
TurtleOnTime copy.py
View File

@ -11,6 +11,14 @@ import mysql_database
from EmailTest import send_email, parse_return_email
from dataclasses import dataclass
import time
'''
todo
1 运行过程框架调整，支持多个turtle同时监测
2 增加运行状态写入yaml文件，读取文件恢复状态
'''
@dataclass
class BuyState:
trigger_time: float # 触发次数
@ -620,110 +628,88 @@ class TurtleTrading_OnTime(object):
Net_return=abs(self.turtle.Capital - available_cash))
self.turtle.tradeslog.append(sale_this_time)
def run_short_trading_loop(self):
while True:
# 获取当前时间
now = datetime.now().time()
def run_short_trading_loop(self, stock_data, etf_data):
# 判断当前时间是否在交易时段内（9:30-11:30 或 13:00-15:00）
is_trading_time = (
(now.hour == 9 and now.minute >= 30) or
(now.hour == 10 and 0 <= now.minute <= 59) or
(now.hour == 11 and now.minute <= 30) or
(now.hour == 13 and 0 <= now.minute <= 59) or
(now.hour == 14 and 0 <= now.minute <= 59) or
(now.hour == 15 and now.minute <= 0)
)
now = datetime.now().time()
# 根据类型获取当前价格
if self.turtle.type == "stock":
self.turtle.PriceNow = float(stock_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
elif self.turtle.type == "etf":
# self.turtle.PriceNow = float(etf_data.loc[etf_data['基金代码'] == self.turtle.TradeCode, '当前-单位净值'].values[0])
self.turtle.PriceNow = float(etf_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
# # 9点30 判断是否跳空高开
if now.hour == 9 and now.minute == 30 and self.turtle.PriceNow > self.turtle.prev_heigh:
self.turtle.is_gap_up = True
# 判断当前仓位状态并执行相应操作
if self.turtle.TrigerTime == 0:
# 空仓状态
if self.turtle.system1EnterNormal(
self.turtle.PriceNow,
self.turtle.Donchian_20_up,
self.turtle.BreakOutLog
):
self.Buy_stock(self.turtle.PriceNow)
# if not is_trading_time:
# # 非交易时间，等待 1 分钟后继续循环
# time.sleep(60)
# continue
# 突破 记录self.turtle.breakoutlog
today = datetime.now().strftime("%Y-%m-%d")
breakout_this_time = BreakOutLog(today,
self.turtle.Donchian_20_up,
self.turtle.Donchian_20_up - 2 * self.turtle.N,
'valid',
None)
self.turtle.BreakOutLog.append(breakout_this_time)
# 获取股票和ETF数据
stock_data, etf_data = self.get_stocks_data()
elif self.turtle.system1EnterSafe(
self.turtle.PriceNow,
self.turtle.Donchian_50_up
):
self.Buy_stock(self.turtle.PriceNow)
# 根据类型获取当前价格
if self.turtle.type == "stock":
self.turtle.PriceNow = float(stock_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
elif self.turtle.type == "etf":
# self.turtle.PriceNow = float(etf_data.loc[etf_data['基金代码'] == self.turtle.TradeCode, '当前-单位净值'].values[0])
self.turtle.PriceNow = float(etf_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
# # 9点30 判断是否跳空高开
if now.hour == 9 and now.minute == 30 and self.turtle.PriceNow > self.turtle.prev_heigh:
self.turtle.is_gap_up = True
elif 1 <= self.turtle.TrigerTime <= 3:
# # 突破状态
# if self.turtle.system1EnterNormal(
# self.turtle.PriceNow,
# self.turtle.Donchian_20_up,
# self.turtle.BreakOutLog
# ):
# self.Buy_stock(self.turtle.PriceNow)
# elif self.turtle.system1EnterSafe(
# self.turtle.PriceNow,
# self.turtle.Donchian_50_up
# ):
# self.Buy_stock(self.turtle.PriceNow)
# 加仓状态
if self.turtle.add(self.turtle.PriceNow):
self.add_stock(self.turtle.PriceNow)
# 判断当前仓位状态并执行相应操作
if self.turtle.TrigerTime == 0:
# 空仓状态
if self.turtle.system1EnterNormal(
self.turtle.PriceNow,
self.turtle.Donchian_20_up,
self.turtle.BreakOutLog
):
self.Buy_stock(self.turtle.PriceNow)
# 突破 记录self.turtle.breakoutlog
today = datetime.now().strftime("%Y-%m-%d")
breakout_this_time = BreakOutLog(today,
self.turtle.Donchian_20_up,
self.turtle.Donchian_20_up - 2 * self.turtle.N,
'valid',
None)
self.turtle.BreakOutLog.append(breakout_this_time)
# 止损状态
elif self.turtle.system_1_stop(self.turtle.PriceNow):
self.stop_sale_stock(self.turtle.PriceNow)
# 止盈
elif self.turtle.system_1_Out(
self.turtle.PriceNow,
self.turtle.Donchian_10_down
):
self.out_sale_stock(self.turtle.PriceNow)
elif self.turtle.system1EnterSafe(
self.turtle.PriceNow,
self.turtle.Donchian_50_up
):
self.Buy_stock(self.turtle.PriceNow)
elif self.turtle.TrigerTime == 4:
# 满仓 止损 止盈
if self.turtle.system_1_stop(self.turtle.PriceNow):
self.stop_sale_stock(self.turtle.PriceNow)
elif self.turtle.system_1_Out(
self.turtle.PriceNow,
self.turtle.Donchian_10_down
):
self.out_sale_stock(self.turtle.PriceNow)
elif 1 <= self.turtle.TrigerTime <= 3:
# # 突破状态
# if self.turtle.system1EnterNormal(
# self.turtle.PriceNow,
# self.turtle.Donchian_20_up,
# self.turtle.BreakOutLog
# ):
# self.Buy_stock(self.turtle.PriceNow)
# elif self.turtle.system1EnterSafe(
# self.turtle.PriceNow,
# self.turtle.Donchian_50_up
# ):
# self.Buy_stock(self.turtle.PriceNow)
# 加仓状态
if self.turtle.add(self.turtle.PriceNow):
self.add_stock(self.turtle.PriceNow)
# 止损状态
elif self.turtle.system_1_stop(self.turtle.PriceNow):
self.stop_sale_stock(self.turtle.PriceNow)
# 止盈
elif self.turtle.system_1_Out(
self.turtle.PriceNow,
self.turtle.Donchian_10_down
):
self.out_sale_stock(self.turtle.PriceNow)
elif self.turtle.TrigerTime == 4:
# 满仓 止损 止盈
if self.turtle.system_1_stop(self.turtle.PriceNow):
self.stop_sale_stock(self.turtle.PriceNow)
elif self.turtle.system_1_Out(
self.turtle.PriceNow,
self.turtle.Donchian_10_down
):
self.out_sale_stock(self.turtle.PriceNow)
# 等待 1 分钟后下一次循环
time.sleep(60)
# 等待 1 分钟后下一次循环
time.sleep(60)
def Start_short_system(self):
"""启动short系统
@ -745,7 +731,34 @@ class TurtleTrading_OnTime(object):
self.turtle.CalPositionSize()
# 每分钟获取一次数据，判断是否触发条件 9:30-11:30 13:00-15:00
self.run_short_trading_loop()
while True:
# 获取当前时间
now = datetime.now().time()
# 判断当前时间是否在交易时段内（9:30-11:30 或 13:00-15:00）
is_trading_time = (
(now.hour == 9 and now.minute >= 30) or
(now.hour == 10 and 0 <= now.minute <= 59) or
(now.hour == 11 and now.minute <= 30) or
(now.hour == 13 and 0 <= now.minute <= 59) or
(now.hour == 14 and 0 <= now.minute <= 59) or
(now.hour == 15 and now.minute <= 0)
)
if not is_trading_time:
# 非交易时间，等待 1 分钟后继续循环
time.sleep(60)
continue
is_stop_time = (now.hour > 15 and now.minute > 0) #收盘时间
if is_stop_time:
break
# 获取股票和ETF数据
stock_data, etf_data = self.get_stocks_data()
self.run_short_trading_loop(stock_data, etf_data)
# ------------------结束阶段--------------------
# 数据库更新当天数据，增加ATR、donchian数据
# 直接做个新表
@ -754,6 +767,7 @@ class TurtleTrading_OnTime(object):
time.sleep(16.5*600)
if __name__ == '__main__':
user_email = "guoyize2209@163.com"
t = TurtleTrading('513870', "etf", 0.0025, 100000, 200000)
# t.get_ready(100)

BIN
__pycache__/data_fetcher.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
__pycache__/turtle_trading.cpython-310.pyc Normal file
View File

Binary file not shown.

59
architecture_plan.md Normal file
View File

@ -0,0 +1,59 @@
# Architecture Plan for TurtleOnTime.py
## Overview
This document outlines a plan to refactor the `TurtleOnTime.py` file to improve its architecture, maintainability, and testability.
## Current Issues
* Single Responsibility Principle violation
* Tight coupling
* Lack of modularity
* Limited testability
## Proposed Solution
1. Refactor the `TurtleTrading` class into smaller, more focused classes.
2. Introduce dependency injection to decouple components.
3. Create a configuration class to store configuration parameters.
4. Implement robust error handling.
5. Add logging.
6. Create unit tests.
## Class Diagram
```mermaid
graph TD
A[Understand Current Architecture] --> B{Read TurtleOnTime.py};
B --> C[Identify Key Components];
C --> D[Analyze Data Flow];
D --> E[Assess Code Structure & Modularity];
E --> F[Identify Potential Bottlenecks & Areas for Improvement];
F --> G[Propose Architectural Changes & Refactoring];
G --> H[Create Detailed Improvement Plan];
H --> I[Present Improvement Plan (Mermaid Diagram)];
I --> J[Seek User Approval];
J -- Approved --> K[Write to File (if approved)];
J -- Rejected --> L[Revise Plan];
L --> I;
```
```mermaid
graph TD
A[TurtleTrading Class] --> B(DataFetcher Class)
A --> C(DonchianChannelCalculator Class)
A --> D(PositionSizer Class)
A --> E(TradingStrategy Class)
F[Config Class] --> A
G[Unit Tests] --> A
```
## Detailed Steps
* **DataFetcher Class:** Responsible for fetching data from various sources (e.g., Akshare). Should be configurable to support different data sources.
* **DonchianChannelCalculator Class:** Calculates Donchian channels. Should be reusable for different timeframes.
* **PositionSizer Class:** Determines the size of positions based on risk tolerance and other factors.
* **TradingStrategy Class:** Encapsulates the trading logic. Should be easily swappable to test different strategies.
* **Config Class:** Stores configuration parameters, such as data source URLs and trading strategy parameters.
* **Error Handling:** Implement try-except blocks to catch and handle exceptions gracefully. Log errors to a file for debugging.
* **Logging:** Use a logging library to track important events and debug issues.
* **Unit Tests:** Create unit tests to verify the correctness of individual components.

6
config.py Normal file
View File

@ -0,0 +1,6 @@
class Config:
def __init__(self, symbol):
self.symbol = symbol
def get_symbol(self):
return self.symbol

23
data_fetcher.py Normal file
View File

@ -0,0 +1,23 @@
import ak
class DataFetcher:
def __init__(self, config):
self.config = config
def fetch_stock_data(self):
"""Fetches stock data from Akshare."""
try:
stock_data = ak.stock_zh_a_spot(symbol=self.config.symbol)
return stock_data
except Exception as e:
print(f"Error fetching stock data: {e}")
return None
def fetch_etf_data(self):
"""Fetches ETF data from Akshare."""
try:
etf_data = ak.fund_etf_spot(symbol=self.config.symbol)
return etf_data
except Exception as e:
print(f"Error fetching ETF data: {e}")
return None

6
run.py Normal file
View File

@ -0,0 +1,6 @@
from turtle_trading import TurtleTrading
if __name__ == "__main__":
symbol = "000001" # Example symbol
turtle = TurtleTrading(symbol)
turtle.run()

16
turtle_trading.py Normal file
View File

@ -0,0 +1,16 @@
from data_fetcher import DataFetcher
from config import Config
class TurtleTrading:
def __init__(self, symbol):
self.symbol = symbol
self.config = Config(self.symbol)
self.data_fetcher = DataFetcher(self.config)
def run(self):
stock_data = self.data_fetcher.fetch_stock_data()
if stock_data is not None:
print(f"Fetched stock data for {self.symbol}:")
print(stock_data)
else:
print(f"Failed to fetch stock data for {self.symbol}.")