turtle_quant_1.strategies.helpers.multiprocessing

turtle_quant_1.strategies.helpers.multiprocessing

BaseCache

Bases: ABC

Base class for process-safe caches.

contains abstractmethod

contains(key: str) -> bool

Check if a key exists in the cache.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

@abstractmethod
def contains(self, key: str) -> bool:
    """Check if a key exists in the cache."""
    raise NotImplementedError()

get abstractmethod

get(key: str) -> Any

Get a value from the cache.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

@abstractmethod
def get(self, key: str) -> Any:
    """Get a value from the cache."""
    raise NotImplementedError()

set abstractmethod

set(key: str, value: Any) -> None

Set a value in the cache.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

@abstractmethod
def set(self, key: str, value: Any) -> None:
    """Set a value in the cache."""
    raise NotImplementedError()

FileCache

FileCache()

Process-safe cache using file-based storage with proper locking.

Initialize the cache with a directory for storage.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

def __init__(self):
    """Initialize the cache with a directory for storage."""
    self.cache_dir_path = self._CACHE_DIR_PATH
    self.cache_dir_path.mkdir(parents=True, exist_ok=True)
    logger.debug(f"Using temporary cache directory '{self.cache_dir_path}'...")

    # Use file-based locking for cross-process synchronization
    self.lock_file = self.cache_dir_path / "cache.lock"
    self._local_cache = {}  # In-memory cache for performance
    self._local_lock = threading.Lock()  # Thread-level locking

    # Increment reference count and clear if first
    with self._global_lock():
        ref_count = self._read_ref_count()
        if ref_count == 0:
            self._clear_cache_dir()
            logger.debug(f"Cleared cache directory '{self.cache_dir_path}'")
        self._write_ref_count(ref_count + 1)
        logger.debug(f"Incremented reference count to {ref_count + 1}")

    atexit.register(self._cleanup)

contains

contains(key: str) -> bool

Check if a key exists in the cache.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

def contains(self, key: str) -> bool:
    """Check if a key exists in the cache."""
    # Check local cache first
    with self._local_lock:
        if key in self._local_cache:
            return True

    # Check file cache
    cache_file = self._get_cache_file_path(key)
    return cache_file.exists()

get

get(key: str) -> Any

Get a value from the cache.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

def get(self, key: str) -> Any:
    """Get a value from the cache."""
    # First check local cache
    with self._local_lock:
        if key in self._local_cache:
            return self._local_cache[key]

    # Check file cache
    cache_file = self._get_cache_file_path(key)
    if cache_file.exists():
        lock_fd = None
        try:
            lock_fd = self._acquire_file_lock()
            with open(cache_file, "rb") as f:
                value = pickle.load(f)

            # Update local cache
            with self._local_lock:
                self._local_cache[key] = value

            return value
        except Exception:
            return None
        finally:
            if lock_fd is not None:
                self._release_file_lock(lock_fd)

    return None

set

set(key: str, value: Any) -> None

Set a value in the cache.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

def set(self, key: str, value: Any) -> None:
    """Set a value in the cache."""
    cache_file = self._get_cache_file_path(key)
    lock_fd = None

    try:
        lock_fd = self._acquire_file_lock()

        # Write to file
        with open(cache_file, "wb") as f:
            pickle.dump(value, f)

        # Update local cache
        with self._local_lock:
            self._local_cache[key] = value

    except Exception as e:
        logger.error(f"Failed to cache data for key {key}: {e}")
    finally:
        if lock_fd is not None:
            self._release_file_lock(lock_fd)

SharedMemoryCache

SharedMemoryCache(
    size_mb: int = 1,
    name: str | None = None,
    lock: Lock | None = None,
)

Bases: BaseCache

A dict-like cache using shared_memory + serialization.

:param size_mb: Number of bytes to allocate for shared memory. :param name: Optional name of existing shared memory block (to attach). :param lock: Optional Lock for synchronization. A new one will be created if None.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

def __init__(
    self,
    size_mb: int = 1,
    name: str | None = None,
    lock: LockType | None = None,
):
    """
    :param size_mb: Number of bytes to allocate for shared memory.
    :param name: Optional name of existing shared memory block (to attach).
    :param lock: Optional Lock for synchronization. A new one will be created if None.
    """
    self.size = size_mb * 1024 * 1024
    self.lock = lock or Lock()

    if name is None:
        # Create new shared memory block
        self.shm = shared_memory.SharedMemory(create=True, size=self.size)
        # Initialize with empty dict
        self._write_dict({})
    else:
        # Attach to existing shared memory block
        self.shm = shared_memory.SharedMemory(name=name)

name property

name: str

Return the name of the underlying shared memory block.

detach

detach()

Detaches the current Python process from the shared memory block.

NOTE: This should be called in each process.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

def detach(self):
    """Detaches the current Python process from the shared memory block.

    NOTE: This should be called in each process.
    """
    self.shm.close()

free

free()

Free the shared memory block (only once, after all processes are done).

NOTE: This should be called only once, after all processes are done.

Source code in src/turtle_quant_1/strategies/helpers/multiprocessing.py

def free(self):
    """Free the shared memory block (only once, after all processes are done).

    NOTE: This should be called only once, after all processes are done.
    """
    self.shm.unlink()