Predicting whether a stock price will rise or fall the next day is a classic example of a **binary classification** problem in machine learning. In this context, you assign a label—such as 1 for "up" and 0 for "down"—to each day based on whether the stock's closing price increases or decreases compared to the previous day. By using features like the previous day's return and moving averages, you can train a model to recognize patterns that may indicate future price movements. Logistic regression, a widely used algorithm for binary classification, is well-suited for this task because it estimates the probability of a binary outcome based on input features. Evaluating the model's accuracy on unseen data helps you understand how well it can generalize to future stock movements, which is crucial for making informed decisions in finance.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression

class TestTask(unittest.TestCase):
    def setUp(self):
        self.prices = [
            100.0, 101.5, 102.0, 101.0, 102.5, 103.0, 102.8,
            104.0, 105.5, 104.5, 106.0, 105.0, 106.5, 107.0
        ]
        self.module = importlib.import_module('user_code')

    def test_prev_return_feature(self):
        # Check that Prev_Return is calculated correctly
        func = getattr(self.module, 'predict_stock_movement', None)
        _dynamic_test(self, callable(func), "Function exists", "Function predict_stock_movement not found")
        # Reproduce expected returns
        df = pd.DataFrame({'Close': self.prices})
        df['Prev_Return'] = df['Close'].pct_change()
        df['MA_3'] = df['Close'].rolling(window=3).mean()
        df['Target'] = (df['Close'].shift(-1) > df['Close']).astype(int)
        df = df.dropna()
        # Call user function and capture feature matrix
        # We'll monkeypatch train_test_split to return features
        orig_train_test_split = self.module.train_test_split
        def fake_split(X, y, test_size, shuffle):
            return X, X, y, y
        self.module.train_test_split = fake_split
        orig_LogisticRegression = self.module.LogisticRegression
        class DummyModel:
            def fit(self, X, y):
                pass
            def predict(self, X):
                return np.zeros(len(X))
        self.module.LogisticRegression = lambda : DummyModel()
        # Run function, catch features
        try:
            self.module.predict_stock_movement(self.prices)
            # Re-create the features in user_code
            X = df[['Prev_Return', 'MA_3']]
            user_X = X.copy()
            _dynamic_test(
                self,
                np.allclose(user_X['Prev_Return'].values, df['Prev_Return'].values, equal_nan=True),
                "Previous day's return feature calculated correctly",
                f"Previous day's return feature mismatch: {user_X['Prev_Return'].values} vs {df['Prev_Return'].values}"
            )
        finally:
            self.module.train_test_split = orig_train_test_split
            self.module.LogisticRegression = orig_LogisticRegression

    def test_ma3_feature(self):
        # Check that MA_3 is calculated correctly
        df = pd.DataFrame({'Close': self.prices})
        df['Prev_Return'] = df['Close'].pct_change()
        df['MA_3'] = df['Close'].rolling(window=3).mean()
        df['Target'] = (df['Close'].shift(-1) > df['Close']).astype(int)
        df = df.dropna()
        X = df[['Prev_Return', 'MA_3']]
        _dynamic_test(
            self,
            np.allclose(X['MA_3'].values, df['MA_3'].values, equal_nan=True),
            "3-day moving average feature calculated correctly",
            f"3-day moving average feature mismatch: {X['MA_3'].values} vs {df['MA_3'].values}"
        )

    def test_target_variable(self):
        df = pd.DataFrame({'Close': self.prices})
        df['Target'] = (df['Close'].shift(-1) > df['Close']).astype(int)
        df['Prev_Return'] = df['Close'].pct_change()
        df['MA_3'] = df['Close'].rolling(window=3).mean()
        df = df.dropna()
        y = df['Target']
        # Check that target is correct
        expected = ((df['Close'].shift(-1) > df['Close']).astype(int)).dropna().values
        _dynamic_test(
            self,
            np.array_equal(y.values, expected),
            "Target variable calculated correctly",
            f"Target variable mismatch: {y.values} vs {expected}"
        )

    def test_drops_nan_rows(self):
        # Check that user drops rows with NaN
        df = pd.DataFrame({'Close': self.prices})
        df['Prev_Return'] = df['Close'].pct_change()
        df['MA_3'] = df['Close'].rolling(window=3).mean()
        df['Target'] = (df['Close'].shift(-1) > df['Close']).astype(int)
        df = df.dropna()
        expected_len = len(df)
        func = getattr(self.module, 'predict_stock_movement', None)
        _dynamic_test(self, callable(func), "Function exists for dropna test", "Function not found for dropna test")
        # Monkeypatch train_test_split to intercept features
        orig_train_test_split = self.module.train_test_split
        def fake_split(X, y, test_size, shuffle):
            return X, X, y, y
        self.module.train_test_split = fake_split
        orig_LogisticRegression = self.module.LogisticRegression
        class DummyModel:
            def fit(self, X, y):
                pass
            def predict(self, X):
                return np.zeros(len(X))
        self.module.LogisticRegression = lambda : DummyModel()
        try:
            self.module.predict_stock_movement(self.prices)
            # Check length after dropna
            _dynamic_test(
                self,
                expected_len == len(df),
                "Rows with NaN values are dropped",
                f"Expected {expected_len} rows after dropna, got {len(df)}"
            )
        finally:
            self.module.train_test_split = orig_train_test_split
            self.module.LogisticRegression = orig_LogisticRegression

    def test_data_split(self):
        # Check that data is split 80% train, 20% test, no shuffle
        func = getattr(self.module, 'predict_stock_movement', None)
        _dynamic_test(self, callable(func), "Function exists for split test", "Function not found for split test")
        df = pd.DataFrame({'Close': self.prices})
        df['Prev_Return'] = df['Close'].pct_change()
        df['MA_3'] = df['Close'].rolling(window=3).mean()
        df['Target'] = (df['Close'].shift(-1) > df['Close']).astype(int)
        df = df.dropna()
        X = df[['Prev_Return', 'MA_3']]
        y = df['Target']
        # Manually split
        n = len(X)
        n_train = int(n * 0.8)
        n_test = n - n_train
        X_train_expected = X.iloc[:n_train]
        X_test_expected = X.iloc[n_train:]
        y_train_expected = y.iloc[:n_train]
        y_test_expected = y.iloc[n_train:]
        # Monkeypatch train_test_split to check order
        orig_train_test_split = self.module.train_test_split
        def fake_split(X, y, test_size, shuffle):
            _dynamic_test(
                self,
                shuffle == False,
                "Data split without shuffling",
                f"train_test_split called with shuffle={shuffle}, expected False"
            )
            n = len(X)
            n_train = int(n * (1 - test_size))
            n_test = n - n_train
            _dynamic_test(
                self,
                np.allclose(X.iloc[:n_train].values, X_train_expected.values),
                "Training set contains first 80% of data",
                f"Training set mismatch: {X.iloc[:n_train].values} vs {X_train_expected.values}"
            )
            _dynamic_test(
                self,
                np.allclose(X.iloc[n_train:].values, X_test_expected.values),
                "Test set contains last 20% of data",
                f"Test set mismatch: {X.iloc[n_train:].values} vs {X_test_expected.values}"
            )
            return X.iloc[:n_train], X.iloc[n_train:], y.iloc[:n_train], y.iloc[n_train:]
        self.module.train_test_split = fake_split
        orig_LogisticRegression = self.module.LogisticRegression
        class DummyModel:
            def fit(self, X, y):
                pass
            def predict(self, X):
                return np.zeros(len(X))
        self.module.LogisticRegression = lambda : DummyModel()
        try:
            self.module.predict_stock_movement(self.prices)
        finally:
            self.module.train_test_split = orig_train_test_split
            self.module.LogisticRegression = orig_LogisticRegression

    def test_model_training_and_accuracy(self):
        # Check that model trains and returns accuracy as float
        func = getattr(self.module, 'predict_stock_movement', None)
        _dynamic_test(self, callable(func), "Function exists for model accuracy", "Function not found for model accuracy")
        acc = func(self.prices)
        _dynamic_test(
            self,
            isinstance(acc, float),
            "Function returns accuracy as float",
            f"Function did not return a float, got {type(acc)}"
        )
        _dynamic_test(
            self,
            0.0 <= acc <= 1.0,
            "Accuracy is between 0 and 1",
            f"Accuracy out of bounds: {acc}"
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()


test_main.py

Explore how Python empowers FinTech solutions by automating financial analysis, modeling, and data-driven decision making. This course blends practical coding with real-world finance scenarios, guiding you through essential FinTech concepts and hands-on challenges.

Learn to analyze, manipulate, and visualize financial data using Python, focusing on core FinTech data workflows.

Apply Python to assess financial risk and optimize investment portfolios using statistical and mathematical techniques.

Apply machine learning techniques to financial data for prediction, classification, and decision support in FinTech applications.

Challenge: Predict Stock Movement

解答